#!/usr/bin/env python3 """ Adaptive Hyperliquid bot. - Trades only BTC, ETH, SOL. - Paper trades by default. - Uses online ML (SGDClassifier) to adapt entry confidence thresholds. - Persists learned state to disk so learning survives restarts. - Can optionally switch to live execution only after consistent paper performance. """ from __future__ import annotations import argparse import json import logging import math import random import os import signal import threading import time from dataclasses import asdict, dataclass from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, Dict, List, Optional, Tuple import joblib import numpy as np import pandas as pd from sklearn.linear_model import SGDClassifier from sklearn.preprocessing import PolynomialFeatures, StandardScaler from hf_advisor import CryptoSentimentAdvisor, HuggingFaceAdvisor from hbot import HyperliquidAPI from telegram_notifier import TelegramNotifier logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) ALLOWED_COINS = ("HYPE",) NUM_FEATURES = 13 # must match build_features() output length FEATURE_NAMES = [ "ret_1", "ret_3", "ret_8", "macd", "atr14", "rsi", "vol_z", "ret_16", "bb_pct", "candle_body", "funding_rate", "oi_change_pct", "htf_trend", ] _SCALER_MIN_SAMPLES = 15 # scaler needs this many samples before transforming _POLY_DEGREE = 2 # degree-2 interaction terms expand 13 raw features → 91 model inputs _VIRTUAL_SAMPLE_WEIGHT = 0.15 # low weight for held-bar synthetic labels VOL_REGIMES = ("low", "normal", "high", "extreme") INTERVAL_MS = { "1m": 60_000, "5m": 300_000, "15m": 900_000, "30m": 1_800_000, "1h": 3_600_000, } @dataclass class Position: coin: str side: str # long | short entry: float size: float sl: float tp: float entry_ts: int entry_features: List[float] entry_p_long: float = 0.5 signal_reason: str = "model" @dataclass class ClosedTrade: coin: str side: str entry: float exit: float size: float pnl_usd: float pnl_pct: float reason: str entry_ts: int exit_ts: int entry_p_long: float = 0.5 class LearningState: """Holds online models + strategy performance and persists to disk.""" def __init__(self, state_path: str): self.state_path = state_path self.feature_count: int = NUM_FEATURES self.models: Dict[str, SGDClassifier] = { c: SGDClassifier( loss="log_loss", alpha=5e-4, # stronger L2 regularisation — reduces regime overfitting random_state=42, ) for c in ALLOWED_COINS } # Per-coin label counts for manual class-balancing (balanced class_weight # is not supported by partial_fit; we replicate it via sample_weight instead). self.label_counts: Dict[str, Dict[int, int]] = {c: {0: 0, 1: 0} for c in ALLOWED_COINS} # Polynomial interaction features: 13 raw → 91 model inputs (degree-2, no bias term). # Stateless transform — fit once to record input shape, then reuse. self.poly = PolynomialFeatures(degree=_POLY_DEGREE, interaction_only=True, include_bias=False) self.poly.fit(np.zeros((1, NUM_FEATURES))) self.model_ready: Dict[str, bool] = {c: False for c in ALLOWED_COINS} self.model_samples: Dict[str, int] = {c: 0 for c in ALLOWED_COINS} self.scalers: Dict[str, StandardScaler] = {c: StandardScaler() for c in ALLOWED_COINS} self.scaler_ready: Dict[str, bool] = {c: False for c in ALLOWED_COINS} self.scaler_samples: Dict[str, int] = {c: 0 for c in ALLOWED_COINS} self.thresholds: Dict[str, float] = {c: 0.56 for c in ALLOWED_COINS} self.trade_stats: Dict[str, Dict[str, float]] = { c: {"trades": 0, "wins": 0, "losses": 0, "gross_profit": 0.0, "gross_loss": 0.0} for c in ALLOWED_COINS } self.ml_stats: Dict[str, Dict[str, float]] = { c: {"count": 0.0, "brier_sum": 0.0} for c in ALLOWED_COINS } self.recent_results: Dict[str, List[int]] = {c: [] for c in ALLOWED_COINS} self.recent_brier: Dict[str, List[float]] = {c: [] for c in ALLOWED_COINS} self.recent_predictions: Dict[str, List[Dict[str, float]]] = {c: [] for c in ALLOWED_COINS} self.feature_history: Dict[str, List[List[float]]] = {c: [] for c in ALLOWED_COINS} self.regime_stats: Dict[str, Dict[str, Dict[str, float]]] = { c: self._default_regime_stats() for c in ALLOWED_COINS } self.closed_trades: List[Dict[str, Any]] = [] self.trade_events: List[Dict[str, Any]] = [] self.equity_curve: List[Dict[str, float]] = [] self.paper_runtime: Dict[str, Any] = { "equity": None, "positions": {}, "entry_features": {}, } self.last_saved_ts = 0.0 self._load() @staticmethod def _default_regime_stats() -> Dict[str, Dict[str, float]]: return { regime: {"trades": 0.0, "wins": 0.0, "pnl_sum": 0.0, "brier_sum": 0.0} for regime in VOL_REGIMES } def _load(self) -> None: if not os.path.exists(self.state_path): logger.info("No prior learning state found at %s, starting fresh.", self.state_path) return payload = joblib.load(self.state_path) saved_feature_count = int(payload.get("feature_count", 7)) saved_poly_degree = int(payload.get("poly_degree", 0)) feat_changed = saved_feature_count != self.feature_count poly_changed = saved_poly_degree != _POLY_DEGREE if feat_changed or poly_changed: logger.warning( "Feature count or poly degree changed (%d/%d -> %d/%d): resetting models but preserving trade history.", saved_feature_count, saved_poly_degree, self.feature_count, _POLY_DEGREE, ) models_compatible = False else: models_compatible = True if models_compatible: # Merge: only load keys for coins that still exist in ALLOWED_COINS. for c in ALLOWED_COINS: saved_models = payload.get("models", {}) if c in saved_models: self.models[c] = saved_models[c] saved_ready = payload.get("model_ready", {}) if c in saved_ready: self.model_ready[c] = saved_ready[c] saved_samples = payload.get("model_samples", {}) if c in saved_samples: self.model_samples[c] = saved_samples[c] saved_brier = payload.get("recent_brier", {}) if c in saved_brier: self.recent_brier[c] = saved_brier[c] saved_ml = payload.get("ml_stats", {}) if c in saved_ml: self.ml_stats[c] = saved_ml[c] saved_scalers = payload.get("scalers", {}) if c in saved_scalers: self.scalers[c] = saved_scalers[c] saved_scaler_ready = payload.get("scaler_ready", {}) if c in saved_scaler_ready: self.scaler_ready[c] = saved_scaler_ready[c] saved_scaler_samples = payload.get("scaler_samples", {}) if c in saved_scaler_samples: self.scaler_samples[c] = saved_scaler_samples[c] saved_label_counts = payload.get("label_counts", {}) if c in saved_label_counts: self.label_counts[c] = saved_label_counts[c] # Merge per-coin dicts so new coins get defaults, old coins don't crash. for c in ALLOWED_COINS: saved_thresh = payload.get("thresholds", {}) if c in saved_thresh: self.thresholds[c] = saved_thresh[c] saved_stats = payload.get("trade_stats", {}) if c in saved_stats: self.trade_stats[c] = saved_stats[c] saved_results = payload.get("recent_results", {}) if c in saved_results: self.recent_results[c] = saved_results[c] saved_predictions = payload.get("recent_predictions", {}) if c in saved_predictions: self.recent_predictions[c] = saved_predictions[c][-500:] saved_feature_history = payload.get("feature_history", {}) if c in saved_feature_history: self.feature_history[c] = saved_feature_history[c][-800:] saved_regime_stats = payload.get("regime_stats", {}) if c in saved_regime_stats and isinstance(saved_regime_stats[c], dict): merged = self._default_regime_stats() for regime, vals in saved_regime_stats[c].items(): if regime in merged and isinstance(vals, dict): merged[regime].update({ "trades": float(vals.get("trades", 0.0)), "wins": float(vals.get("wins", 0.0)), "pnl_sum": float(vals.get("pnl_sum", 0.0)), "brier_sum": float(vals.get("brier_sum", 0.0)), }) self.regime_stats[c] = merged self.closed_trades = payload.get("closed_trades", self.closed_trades) self.trade_events = payload.get("trade_events", self.trade_events) self.equity_curve = payload.get("equity_curve", self.equity_curve) self.paper_runtime = payload.get("paper_runtime", self.paper_runtime) logger.info("Loaded learning state from %s", self.state_path) def save(self, force: bool = False) -> None: now = time.time() if not force and now - self.last_saved_ts < 20: return payload = { "feature_count": self.feature_count, "poly_degree": _POLY_DEGREE, "models": self.models, "model_ready": self.model_ready, "model_samples": self.model_samples, "scalers": self.scalers, "scaler_ready": self.scaler_ready, "scaler_samples": self.scaler_samples, "label_counts": self.label_counts, "thresholds": self.thresholds, "trade_stats": self.trade_stats, "ml_stats": self.ml_stats, "recent_results": self.recent_results, "recent_brier": self.recent_brier, "recent_predictions": self.recent_predictions, "feature_history": self.feature_history, "regime_stats": self.regime_stats, "closed_trades": self.closed_trades[-2000:], "trade_events": self.trade_events[-5000:], "equity_curve": self.equity_curve[-5000:], "paper_runtime": self.paper_runtime, "saved_at": datetime.now(timezone.utc).isoformat(), } joblib.dump(payload, self.state_path) self.last_saved_ts = now def record_equity(self, equity: float, ts_ms: Optional[int] = None) -> None: ts = int(ts_ms if ts_ms is not None else time.time() * 1000) e = float(equity) if self.equity_curve and abs(float(self.equity_curve[-1]["equity"]) - e) < 1e-12: return self.equity_curve.append({"ts": float(ts), "equity": e}) self.equity_curve = self.equity_curve[-5000:] def set_runtime_state( self, paper_equity: float, positions: Dict[str, Position], entry_features: Dict[Tuple[str, int], List[float]], ) -> None: self.paper_runtime = { "equity": float(paper_equity), "positions": {coin: asdict(pos) for coin, pos in positions.items()}, "entry_features": { f"{coin}:{entry_ts}": [float(v) for v in feats] for (coin, entry_ts), feats in entry_features.items() }, } def restore_runtime_state(self) -> Tuple[Optional[float], Dict[str, Position], Dict[Tuple[str, int], List[float]]]: equity_raw = self.paper_runtime.get("equity") equity = float(equity_raw) if equity_raw is not None else None restored_positions: Dict[str, Position] = {} for coin, raw in (self.paper_runtime.get("positions") or {}).items(): try: restored_positions[coin] = Position( coin=str(raw["coin"]), side=str(raw["side"]), entry=float(raw["entry"]), size=float(raw["size"]), sl=float(raw["sl"]), tp=float(raw["tp"]), entry_ts=int(raw["entry_ts"]), entry_features=[float(v) for v in raw.get("entry_features", [])], entry_p_long=float(raw.get("entry_p_long", 0.5)), signal_reason=str(raw.get("signal_reason", "model")), ) except Exception: continue restored_features: Dict[Tuple[str, int], List[float]] = {} for key, vals in (self.paper_runtime.get("entry_features") or {}).items(): try: coin, entry_ts_str = key.split(":", 1) restored_features[(coin, int(entry_ts_str))] = [float(v) for v in vals] except Exception: continue return equity, restored_positions, restored_features def record_trade_event(self, event_type: str, payload: Dict[str, Any]) -> None: self.trade_events.append( { "event": event_type, "timestamp": datetime.now(timezone.utc).isoformat(), **payload, } ) self.trade_events = self.trade_events[-5000:] def update_after_trade(self, trade: ClosedTrade, entry_features: List[float], skip_model_fit: bool = False) -> None: trade_won = 1 if trade.pnl_usd > 0 else 0 # Model predicts p(long wins), so short outcomes must be direction-adjusted. label_long = trade_won if trade.side == "long" else (1 - trade_won) # Pad to NUM_FEATURES in case features were stored with an older (shorter) feature set. feat_padded = list(entry_features[:NUM_FEATURES]) if len(feat_padded) < NUM_FEATURES: feat_padded.extend([0.0] * (NUM_FEATURES - len(feat_padded))) x_raw = np.asarray(feat_padded, dtype=float).reshape(1, -1) y = np.asarray([label_long], dtype=int) # Always update the scaler with new observations (even bootstrap trades). scaler = self.scalers[trade.coin] scaler.partial_fit(x_raw) self.scaler_samples[trade.coin] = int(self.scaler_samples.get(trade.coin, 0)) + 1 if self.scaler_samples[trade.coin] >= _SCALER_MIN_SAMPLES: self.scaler_ready[trade.coin] = True # Only fit the model for non-bootstrap trades so early noisy labels don't pollute it. if not skip_model_fit: x = scaler.transform(x_raw) if self.scaler_ready[trade.coin] else x_raw x = self.poly.transform(x) # Manual class balancing: track label counts and upweight the minority class. counts = self.label_counts[trade.coin] counts[label_long] = counts.get(label_long, 0) + 1 total = counts.get(0, 0) + counts.get(1, 0) balance_weight = float(total) / (2.0 * max(counts[label_long], 1)) # Combine with outcome magnitude: a 10% PnL trade gets 1.0 base weight. pnl_weight = float(np.clip(abs(trade.pnl_pct) / 10.0, 0.1, 3.0)) sw = np.array([pnl_weight * balance_weight]) model = self.models[trade.coin] if not self.model_ready[trade.coin]: model.partial_fit(x, y, classes=np.array([0, 1]), sample_weight=sw) self.model_ready[trade.coin] = True else: model.partial_fit(x, y, sample_weight=sw) self.model_samples[trade.coin] = int(self.model_samples.get(trade.coin, 0)) + 1 stats = self.trade_stats[trade.coin] stats["trades"] += 1 if trade_won == 1: stats["wins"] += 1 stats["gross_profit"] += max(0.0, trade.pnl_usd) else: stats["losses"] += 1 stats["gross_loss"] += abs(min(0.0, trade.pnl_usd)) p = float(np.clip(trade.entry_p_long, 0.0, 1.0)) brier = float((p - label_long) ** 2) ml = self.ml_stats[trade.coin] ml["count"] += 1 ml["brier_sum"] += brier self.recent_brier[trade.coin].append(brier) self.recent_brier[trade.coin] = self.recent_brier[trade.coin][-100:] self.recent_results[trade.coin].append(trade_won) self.recent_results[trade.coin] = self.recent_results[trade.coin][-30:] self.recent_predictions[trade.coin].append( { "p_long": p, "label": float(label_long), "ts": float(trade.exit_ts), } ) self.recent_predictions[trade.coin] = self.recent_predictions[trade.coin][-500:] feat_vec = [float(v) for v in entry_features[:NUM_FEATURES]] if len(feat_vec) < NUM_FEATURES: feat_vec.extend([0.0] * (NUM_FEATURES - len(feat_vec))) self.feature_history[trade.coin].append(feat_vec) self.feature_history[trade.coin] = self.feature_history[trade.coin][-800:] regime = self.classify_regime(feat_vec) bucket = self.regime_stats[trade.coin].setdefault( regime, {"trades": 0.0, "wins": 0.0, "pnl_sum": 0.0, "brier_sum": 0.0}, ) bucket["trades"] += 1.0 bucket["wins"] += float(trade_won) bucket["pnl_sum"] += float(trade.pnl_usd) bucket["brier_sum"] += float(brier) # Adaptive threshold: lower when recent win rate is strong, raise when weak. recent = self.recent_results[trade.coin] if recent: wr = float(sum(recent)) / len(recent) target = 0.56 + (0.60 - wr) * 0.25 self.thresholds[trade.coin] = float(np.clip(target, 0.52, 0.65)) # Feature drift response: widen threshold when distribution shifts significantly. drift = self._feature_drift_for_coin(trade.coin) if drift is not None and drift["score"] > 1.5: widen = float(np.clip((drift["score"] - 1.5) * 0.03, 0.0, 0.05)) self.thresholds[trade.coin] = float(np.clip( self.thresholds[trade.coin] + widen, 0.52, 0.70 )) logger.info( "%s drift-threshold widen: drift_score=%.2f new_threshold=%.3f", trade.coin, drift["score"], self.thresholds[trade.coin], ) self.closed_trades.append(asdict(trade)) self.save() def train_virtual_sample(self, coin: str, features: List[float], label: int) -> None: """Train on a held-bar observation: prev features → next-bar direction. These are low-weight synthetic labels that massively increase sample throughput (one per poll cycle instead of one per closed trade). Only called after the model has already been initialised from real trades so direction bias can't form. """ if not self.model_ready.get(coin): return feat_padded = list(features[:NUM_FEATURES]) if len(feat_padded) < NUM_FEATURES: feat_padded.extend([0.0] * (NUM_FEATURES - len(feat_padded))) x_raw = np.asarray(feat_padded, dtype=float).reshape(1, -1) x = self.scalers[coin].transform(x_raw) if self.scaler_ready.get(coin) else x_raw x = self.poly.transform(x) sw = np.array([_VIRTUAL_SAMPLE_WEIGHT]) self.models[coin].partial_fit(x, np.array([label]), sample_weight=sw) def enough_samples(self, coin: str, min_samples: int) -> bool: return int(self.model_samples.get(coin, 0)) >= int(max(1, min_samples)) def classify_regime(self, features: List[float]) -> str: atr14 = float(features[4]) if len(features) > 4 else 0.0 vol_z = float(features[6]) if len(features) > 6 else 0.0 if atr14 >= 0.03 or vol_z >= 2.0: return "extreme" if atr14 >= 0.018 or vol_z >= 1.0: return "high" if atr14 <= 0.008 and vol_z <= -0.25: return "low" return "normal" @staticmethod def _ece(samples: List[Dict[str, float]], bins: int = 5) -> Optional[float]: if not samples: return None total = len(samples) ece = 0.0 for i in range(bins): lo = i / bins hi = (i + 1) / bins if i == bins - 1: bucket = [s for s in samples if lo <= s["p_long"] <= hi] else: bucket = [s for s in samples if lo <= s["p_long"] < hi] if not bucket: continue avg_p = float(np.mean([s["p_long"] for s in bucket])) avg_y = float(np.mean([s["label"] for s in bucket])) ece += (len(bucket) / total) * abs(avg_p - avg_y) return float(ece) def _feature_drift_for_coin(self, coin: str, recent_window: int = 40, baseline_window: int = 240) -> Optional[Dict[str, Any]]: history = self.feature_history.get(coin, []) if len(history) < (recent_window + 40): return None recent = np.asarray(history[-recent_window:], dtype=float) baseline_start = max(0, len(history) - recent_window - baseline_window) baseline = np.asarray(history[baseline_start:-recent_window], dtype=float) if len(baseline) < 40: return None mean_recent = recent.mean(axis=0) mean_base = baseline.mean(axis=0) std_base = baseline.std(axis=0) z = np.abs(mean_recent - mean_base) / np.maximum(std_base, 1e-6) top_idx = int(np.argmax(z)) drift_score = float(np.mean(z)) return { "score": drift_score, "max_feature": FEATURE_NAMES[top_idx], "max_feature_z": float(z[top_idx]), "samples_recent": int(len(recent)), "samples_baseline": int(len(baseline)), } def regime_performance(self, coin: str, regime: str) -> Dict[str, Optional[float]]: regimes = self.regime_stats.get(coin, {}) bucket = regimes.get(regime, {}) trades = int(bucket.get("trades", 0.0)) wins = float(bucket.get("wins", 0.0)) pnl_sum = float(bucket.get("pnl_sum", 0.0)) brier_sum = float(bucket.get("brier_sum", 0.0)) if trades <= 0: return { "regime": regime, "trades": 0, "win_rate": None, "avg_pnl_usd": None, "brier": None, } return { "regime": regime, "trades": trades, "win_rate": wins / trades, "avg_pnl_usd": pnl_sum / trades, "brier": brier_sum / trades, } def predict_long_probability(self, coin: str, features: List[float]) -> float: if not self.model_ready.get(coin, False): return 0.50 x = np.asarray(features, dtype=float).reshape(1, -1) try: if self.scaler_ready.get(coin, False): x = self.scalers[coin].transform(x) x = self.poly.transform(x) p = float(self.models[coin].predict_proba(x)[0][1]) except Exception: return 0.50 if math.isnan(p): return 0.50 return float(np.clip(p, 0.01, 0.99)) def global_metrics(self) -> Dict[str, float]: trades = wins = 0 gross_profit = gross_loss = 0.0 for coin in ALLOWED_COINS: s = self.trade_stats[coin] trades += int(s["trades"]) wins += int(s["wins"]) gross_profit += float(s["gross_profit"]) gross_loss += float(s["gross_loss"]) win_rate = (wins / trades) if trades else 0.0 profit_factor = (gross_profit / gross_loss) if gross_loss > 0 else (10.0 if gross_profit > 0 else 0.0) return { "trades": float(trades), "win_rate": float(win_rate), "profit_factor": float(profit_factor), "gross_profit": float(gross_profit), "gross_loss": float(gross_loss), } def ml_progress_metrics(self) -> Dict[str, Any]: total_count = 0.0 total_brier_sum = 0.0 per_coin: Dict[str, Any] = {} all_recent_brier: List[float] = [] all_recent_preds: List[Dict[str, float]] = [] drift_scores: List[float] = [] for coin in ALLOWED_COINS: stat = self.ml_stats.get(coin, {}) cnt = float(stat.get("count", 0.0)) bsum = float(stat.get("brier_sum", 0.0)) total_count += cnt total_brier_sum += bsum recent = [float(v) for v in self.recent_brier.get(coin, [])] all_recent_brier.extend(recent) preds = [ { "p_long": float(r.get("p_long", 0.5)), "label": float(r.get("label", 0.0)), } for r in self.recent_predictions.get(coin, []) ] all_recent_preds.extend(preds[-100:]) pred_window = preds[-50:] ece_50 = self._ece(pred_window, bins=5) edge_50 = ( float(np.mean([(2.0 * p["label"] - 1.0) * (2.0 * p["p_long"] - 1.0) for p in pred_window])) if pred_window else None ) directional_acc_50 = ( float(np.mean([((p["p_long"] >= 0.5) == (p["label"] >= 0.5)) for p in pred_window])) if pred_window else None ) drift = self._feature_drift_for_coin(coin) if drift is not None: drift_scores.append(float(drift["score"])) regime_perf = {regime: self.regime_performance(coin, regime) for regime in VOL_REGIMES} per_coin[coin] = { "samples": int(cnt), "brier": (bsum / cnt) if cnt > 0 else None, "brier_rolling_50": (float(np.mean(recent[-50:])) if recent else None), "brier_rolling_100": (float(np.mean(recent)) if recent else None), "calibration_error_rolling_50": ece_50, "edge_rolling_50": edge_50, "directional_accuracy_rolling_50": directional_acc_50, "feature_drift": drift, "regime_performance": regime_perf, } brier_global = (total_brier_sum / total_count) if total_count > 0 else None brier_rolling_50 = float(np.mean(all_recent_brier[-50:])) if all_recent_brier else None brier_rolling_100 = float(np.mean(all_recent_brier[-100:])) if all_recent_brier else None calibration_error_rolling_50 = self._ece(all_recent_preds[-50:], bins=5) edge_rolling_50 = ( float(np.mean([(2.0 * p["label"] - 1.0) * (2.0 * p["p_long"] - 1.0) for p in all_recent_preds[-50:]])) if all_recent_preds else None ) directional_accuracy_rolling_50 = ( float(np.mean([((p["p_long"] >= 0.5) == (p["label"] >= 0.5)) for p in all_recent_preds[-50:]])) if all_recent_preds else None ) pnl = [float(t.get("pnl_usd", 0.0)) for t in self.closed_trades] expectancy_30 = float(np.mean(pnl[-30:])) if pnl else None expectancy_100 = float(np.mean(pnl[-100:])) if pnl else None max_drawdown_pct = None peak = None max_dd = 0.0 for point in self.equity_curve: eq = float(point.get("equity", 0.0)) if peak is None or eq > peak: peak = eq if peak and peak > 0: dd = (peak - eq) / peak if dd > max_dd: max_dd = dd if peak is not None: max_drawdown_pct = max_dd * 100.0 return { "samples": int(total_count), "brier": brier_global, "brier_rolling_50": brier_rolling_50, "brier_rolling_100": brier_rolling_100, "calibration_error_rolling_50": calibration_error_rolling_50, "edge_rolling_50": edge_rolling_50, "directional_accuracy_rolling_50": directional_accuracy_rolling_50, "expectancy_usd_30": expectancy_30, "expectancy_usd_100": expectancy_100, "max_drawdown_pct": max_drawdown_pct, "feature_drift_score": (float(np.mean(drift_scores)) if drift_scores else None), "per_coin": per_coin, } def consistently_profitable(self) -> bool: m = self.global_metrics() if not (m["trades"] >= 50 and m["win_rate"] >= 0.57 and m["profit_factor"] >= 1.30): return False # Require the model shows genuine predictive value (Brier < 0.23; random = 0.25). ml = self.ml_progress_metrics() brier = ml.get("brier_rolling_50") if brier is not None and brier > 0.23: return False # Require positive recent expectancy over last 20 trades. exp = ml.get("expectancy_usd_30") if exp is not None and exp <= 0.0: return False return True class PaperBroker: def __init__(self, starting_equity: float = 10000.0, taker_fee_bps: float = 4.32): self.equity = float(starting_equity) self.taker_fee = taker_fee_bps / 10_000.0 self.positions: Dict[str, Position] = {} self.closed: List[ClosedTrade] = [] def has_position(self, coin: str) -> bool: return coin in self.positions def open_position(self, pos: Position) -> None: self.positions[pos.coin] = pos def force_close(self, coin: str, exit_px: float, now_ts: int, reason: str) -> Optional[ClosedTrade]: pos = self.positions.get(coin) if not pos: return None gross = (exit_px - pos.entry) * pos.size if pos.side == "long" else (pos.entry - exit_px) * pos.size fees = (pos.entry * pos.size + exit_px * pos.size) * self.taker_fee pnl = gross - fees pnl_pct = (pnl / max(pos.entry * pos.size, 1e-9)) * 100.0 trade = ClosedTrade( coin=coin, side=pos.side, entry=pos.entry, exit=exit_px, size=pos.size, pnl_usd=pnl, pnl_pct=pnl_pct, reason=reason, entry_ts=pos.entry_ts, exit_ts=now_ts, entry_p_long=pos.entry_p_long, ) self.equity += pnl self.closed.append(trade) del self.positions[coin] return trade def maybe_close(self, coin: str, candle: Dict[str, float], now_ts: int) -> Optional[ClosedTrade]: pos = self.positions.get(coin) if not pos: return None high = candle["high"] low = candle["low"] close = candle["close"] exit_px = close reason = "model_exit" if pos.side == "long": if low <= pos.sl: exit_px = pos.sl reason = "stop_loss" elif high >= pos.tp: exit_px = pos.tp reason = "take_profit" else: if high >= pos.sl: exit_px = pos.sl reason = "stop_loss" elif low <= pos.tp: exit_px = pos.tp reason = "take_profit" if reason == "model_exit": return None gross = (exit_px - pos.entry) * pos.size if pos.side == "long" else (pos.entry - exit_px) * pos.size fees = (pos.entry * pos.size + exit_px * pos.size) * self.taker_fee pnl = gross - fees pnl_pct = (pnl / max(pos.entry * pos.size, 1e-9)) * 100.0 trade = ClosedTrade( coin=coin, side=pos.side, entry=pos.entry, exit=exit_px, size=pos.size, pnl_usd=pnl, pnl_pct=pnl_pct, reason=reason, entry_ts=pos.entry_ts, exit_ts=now_ts, entry_p_long=pos.entry_p_long, ) self.equity += pnl self.closed.append(trade) del self.positions[coin] return trade def parse_candles(raw: List[Dict[str, Any]]) -> pd.DataFrame: rows = [] for c in raw: rows.append( { "ts": int(c.get("t") or c.get("time") or c.get("T") or 0), "open": float(c.get("o") or c.get("open") or c.get("openPx") or 0.0), "high": float(c.get("h") or c.get("high") or c.get("highPx") or 0.0), "low": float(c.get("l") or c.get("low") or c.get("lowPx") or 0.0), "close": float(c.get("c") or c.get("close") or c.get("closePx") or 0.0), "volume": float(c.get("v") or c.get("volume") or 0.0), } ) df = pd.DataFrame(rows).sort_values("ts") df = df[(df["close"] > 0) & (df["high"] > 0) & (df["low"] > 0)] return df.reset_index(drop=True) def build_features( df: pd.DataFrame, funding_rate: float = 0.0, oi_change_pct: float = 0.0, htf_trend: float = 0.0, ) -> Optional[List[float]]: if len(df) < 60: return None close = df["close"] open_ = df["open"] high = df["high"] low = df["low"] vol = df["volume"].replace(0, np.nan).ffill().fillna(1.0) ret_1 = close.pct_change(1) ret_3 = close.pct_change(3) ret_8 = close.pct_change(8) ret_16 = close.pct_change(16) ema_12 = close.ewm(span=12).mean() ema_26 = close.ewm(span=26).mean() macd = (ema_12 - ema_26) / close tr = pd.concat( [ (high - low).abs(), (high - close.shift(1)).abs(), (low - close.shift(1)).abs(), ], axis=1, ).max(axis=1) atr14 = tr.rolling(14).mean() / close delta = close.diff() up = delta.clip(lower=0).rolling(14).mean() down = (-delta.clip(upper=0)).rolling(14).mean() rs = up / down.replace(0, np.nan) rsi = 100.0 - (100.0 / (1.0 + rs)) rsi_scaled = (rsi - 50.0) / 50.0 vol_z = (np.log(vol) - np.log(vol).rolling(20).mean()) / np.log(vol).rolling(20).std() # Bollinger Band %B: 0=at lower band, 1=at upper band, 0.5=at midline. bb_mid = close.rolling(20).mean() bb_std = close.rolling(20).std() bb_pct = (close - (bb_mid - 2 * bb_std)) / (4 * bb_std.replace(0, np.nan)) # Candle body fraction: how much of the bar's range was directional. bar_range = (high - low).replace(0, np.nan) candle_body = (close - open_).abs() / bar_range feat = pd.DataFrame( { "ret_1": ret_1, "ret_3": ret_3, "ret_8": ret_8, "macd": macd, "atr14": atr14, "rsi": rsi_scaled, "vol_z": vol_z, "ret_16": ret_16, "bb_pct": bb_pct, "candle_body": candle_body, } ).replace([np.inf, -np.inf], np.nan) last = feat.iloc[-1].fillna(0.0) base = [ float(last[k]) for k in ["ret_1", "ret_3", "ret_8", "macd", "atr14", "rsi", "vol_z", "ret_16", "bb_pct", "candle_body"] ] # Perp-specific and trend features — clipped to avoid polluting the scaler. # Funding: raw HL value is ~0.0001 per 8h; * 1000 puts it in a model-friendly range. f_rate = float(np.clip(float(funding_rate) * 1000.0, -5.0, 5.0)) oi_chg = float(np.clip(float(oi_change_pct), -0.2, 0.2)) htf = float(np.clip(float(htf_trend), -3.0, 3.0)) return base + [f_rate, oi_chg, htf] def atr_for_risk(df: pd.DataFrame) -> float: high = df["high"] low = df["low"] close = df["close"] tr = pd.concat( [ (high - low).abs(), (high - close.shift(1)).abs(), (low - close.shift(1)).abs(), ], axis=1, ).max(axis=1) atr = float(tr.rolling(14).mean().iloc[-1]) if not np.isfinite(atr) or atr <= 0: atr = float(close.iloc[-1] * 0.01) return atr class AdaptiveHyperliquidBot: def __init__( self, interval: str, state_file: str, report_file: str, starting_equity: float, risk_per_trade: float, poll_seconds: int, exploration_rate: float, enable_live: bool, require_live_approval: bool, live_approval_mode: str, live_approval_file: str, live_startup_test: bool, live_startup_test_coin: str, live_startup_test_notional: float, hf_advisor_url: str, hf_advisor_token: str, hf_advisor_mode: str, hf_advisor_min_confidence: float, hf_advisor_blend_weight: float, hf_advisor_timeout_seconds: float, min_model_samples_for_signal: int, pause_on_negative_expectancy: bool, pause_expectancy_lookback: int, pause_expectancy_threshold_usd: float, pause_max_drawdown_pct: float, pause_cooldown_minutes: float, telegram_ml_report_hours: float, telegram_ml_report_min_trades: int, health_host: str, health_port: int, sl_mult: float, tp_mult: float, risk_volatility_floor_mult: float, risk_volatility_cap_mult: float, risk_regime_adapt_strength: float, risk_regime_min_samples: int, ): self.interval = interval self.poll_seconds = poll_seconds self.risk_per_trade = risk_per_trade self.exploration_rate = float(np.clip(exploration_rate, 0.0, 0.5)) self.enable_live = enable_live self.require_live_approval = require_live_approval self.live_approval_mode = live_approval_mode self.live_approval_file = live_approval_file self.live_startup_test = live_startup_test self.live_startup_test_coin = live_startup_test_coin self.live_startup_test_notional = float(max(10.0, live_startup_test_notional)) self.hf_advisor_mode = (hf_advisor_mode or "blend").strip().lower() if self.hf_advisor_mode not in {"confirm", "veto", "blend"}: logger.warning("Unknown advisor mode '%s'; defaulting to 'blend'.", self.hf_advisor_mode) self.hf_advisor_mode = "blend" self.hf_advisor_min_confidence = float(np.clip(hf_advisor_min_confidence, 0.0, 1.0)) self.hf_advisor_blend_weight = float(np.clip(hf_advisor_blend_weight, 0.0, 1.0)) _hf_token = os.environ.get("HF_TOKEN", "").strip() if _hf_token: self.hf_advisor: Any = CryptoSentimentAdvisor( hf_token=_hf_token, timeout_seconds=hf_advisor_timeout_seconds, cache_seconds=300.0, ) # Auto-enable blending at 20% when sentiment advisor is active and user # hasn't explicitly set a non-zero weight. if self.hf_advisor_blend_weight == 0.0: self.hf_advisor_blend_weight = 0.20 logger.info( "CryptoSentimentAdvisor active (Reddit + Fear&Greed + CryptoBERT). blend_weight=%.2f", self.hf_advisor_blend_weight, ) else: self.hf_advisor = HuggingFaceAdvisor( endpoint_url=hf_advisor_url, api_token=hf_advisor_token, timeout_seconds=hf_advisor_timeout_seconds, ) logger.info("HF_TOKEN not set — using HuggingFaceAdvisor (local heuristic).") self.min_model_samples_for_signal = int(max(10, min_model_samples_for_signal)) self.pause_on_negative_expectancy = bool(pause_on_negative_expectancy) self.pause_expectancy_lookback = int(max(10, pause_expectancy_lookback)) self.pause_expectancy_threshold_usd = float(pause_expectancy_threshold_usd) self.pause_max_drawdown_pct = float(max(0.0, pause_max_drawdown_pct)) self.pause_cooldown_minutes = float(max(0.0, pause_cooldown_minutes)) self._safety_pause_started_ts: Optional[float] = None self._safety_pause_wait_recovery: bool = False self.telegram_ml_report_hours = float(max(0.0, telegram_ml_report_hours)) self.telegram_ml_report_min_trades = int(max(0, telegram_ml_report_min_trades)) self.health_host = health_host self.health_port = int(health_port) self.sl_mult = float(max(0.1, sl_mult)) self.tp_mult = float(max(0.1, tp_mult)) self.risk_volatility_floor_mult = float(np.clip(risk_volatility_floor_mult, 0.2, 1.0)) self.risk_volatility_cap_mult = float(np.clip(risk_volatility_cap_mult, 1.0, 2.0)) self.risk_regime_adapt_strength = float(np.clip(risk_regime_adapt_strength, 0.0, 1.0)) self.risk_regime_min_samples = int(max(5, risk_regime_min_samples)) self.api = HyperliquidAPI(verify_ssl=True) self.learning = LearningState(state_file) self.paper = PaperBroker(starting_equity=starting_equity) self.report_file = report_file self.notifier = TelegramNotifier(disable_on_error=True) self.notifications_enabled = bool( os.environ.get("TELEGRAM_BOT_TOKEN", "").strip() and os.environ.get("TELEGRAM_CHAT_ID", "").strip() and os.environ.get("PROXY_URL", "").strip() ) self._last_summary_date: Optional[str] = None if self.notifications_enabled: self.notifier.send_info("Adaptive ML bot started. Trade notifications are enabled.") else: logger.info("Telegram notifications disabled (need TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, PROXY_URL).") self._running = True self._live_clients: Optional[Dict[str, Any]] = None self._live_readiness_alerted = False self._live_startup_test_done = False self._live_startup_test_passed: Optional[bool] = None self._http_server: Optional[ThreadingHTTPServer] = None self._entry_features: Dict[Tuple[str, int], List[float]] = {} self._bootstrap_entries: set = set() # tracks entry keys that were bootstrap trades self._prev_oi: Dict[str, float] = {} self._asset_ctx_cache: Dict[str, Any] = {} # {coin: (ts, (funding, oi_chg))} self._prev_bar_features: Dict[str, Optional[List[float]]] = {} self._prev_bar_close: Dict[str, Optional[float]] = {} self._consec_loss_pause_ts: Optional[float] = None self._last_ml_report_ts: float = 0.0 self._last_stats_log_ts: float = 0.0 self._last_hold_log_ts: Dict[str, float] = {} self._advisor_stats: Dict[str, int] = { "requests": 0, "errors": 0, "remote_used": 0, "local_used": 0, "held_by_advisor": 0, "kept_model_signal": 0, "used_advisor_blend": 0, } self._last_risk_snapshot: Dict[str, Any] = {} self._restore_runtime_state() def _interval_ms(self) -> int: if self.interval not in INTERVAL_MS: raise ValueError(f"Unsupported interval '{self.interval}'. Supported: {sorted(INTERVAL_MS)}") return INTERVAL_MS[self.interval] def _fetch_market_context(self, coin: str) -> Tuple[float, float]: """Return (funding_rate, oi_change_pct) for coin, cached for 60s.""" now = time.time() cached = self._asset_ctx_cache.get(coin) if cached is not None and (now - cached[0]) < 60.0: return cached[1] ctx = self.api.get_asset_ctx(coin) if ctx is None: return 0.0, 0.0 funding_rate = float(ctx.get("funding", 0.0)) oi = float(ctx.get("openInterest", 0.0)) prev_oi = self._prev_oi.get(coin, oi) oi_change_pct = float(np.clip((oi - prev_oi) / max(abs(prev_oi), 1e-9), -0.2, 0.2)) self._prev_oi[coin] = oi result = (funding_rate, oi_change_pct) self._asset_ctx_cache[coin] = (now, result) return result def _compute_htf_trend(self, df: pd.DataFrame) -> float: """HTF trend as (close - EMA50) / ATR14, in volatility units, clipped to [-3, 3].""" close = df["close"] ema50 = float(close.ewm(span=50, adjust=False).mean().iloc[-1]) atr = atr_for_risk(df) if atr <= 0 or not np.isfinite(ema50): return 0.0 return float(np.clip((float(close.iloc[-1]) - ema50) / atr, -3.0, 3.0)) def stop(self, *_: Any) -> None: logger.info("Received shutdown signal, flushing learning state.") self._running = False if self._http_server is not None: self._http_server.shutdown() self._http_server.server_close() self._http_server = None self._persist_runtime_state() self.learning.save(force=True) self.write_report() def status_payload(self) -> Dict[str, Any]: metrics = self.learning.global_metrics() approval = self._live_approval_status_details() safety_pause = { "enabled": self.pause_on_negative_expectancy, "lookback": self.pause_expectancy_lookback, "expectancy_threshold_usd": self.pause_expectancy_threshold_usd, "max_drawdown_pct": self.pause_max_drawdown_pct, "cooldown_minutes": self.pause_cooldown_minutes, "active": self._safety_pause_started_ts is not None, "active_since_ts": self._safety_pause_started_ts, "wait_recovery": self._safety_pause_wait_recovery, } return { "service": "hyperliquid-adaptive-bot", "status": "ok" if self._running else "stopping", "timestamp": datetime.now(timezone.utc).isoformat(), "paper_equity": self.paper.equity, "open_positions": [asdict(p) for p in self.paper.positions.values()], "metrics": metrics, "ml_progress": self.learning.ml_progress_metrics(), "thresholds": self.learning.thresholds, "state_file": self.learning.state_path, "consistently_profitable": self.learning.consistently_profitable(), "live_armed": self._live_clients is not None, "live_approval_required": self.require_live_approval, "live_approval_mode": self.live_approval_mode, "live_approval_granted": approval["granted"], "live_approval_checks": approval, "live_startup_test_enabled": self.live_startup_test, "live_startup_test_done": self._live_startup_test_done, "live_startup_test_passed": self._live_startup_test_passed, "trade_events_recorded": len(self.learning.trade_events), "advisor": { "enabled": self.hf_advisor.enabled, "remote_enabled": self.hf_advisor.remote_enabled, "mode": self.hf_advisor_mode, "min_confidence": self.hf_advisor_min_confidence, "blend_weight": self.hf_advisor_blend_weight, "stats": self._advisor_stats, }, "model_samples": self.learning.model_samples, "min_model_samples_for_signal": self.min_model_samples_for_signal, "safety_pause": safety_pause, "risk_sizing": { "base_risk_per_trade": self.risk_per_trade, "volatility_floor_mult": self.risk_volatility_floor_mult, "volatility_cap_mult": self.risk_volatility_cap_mult, "regime_adapt_strength": self.risk_regime_adapt_strength, "regime_min_samples": self.risk_regime_min_samples, "last_snapshot": self._last_risk_snapshot, }, } def status_overview_payload(self) -> Dict[str, Any]: metrics = self.learning.global_metrics() ml_progress = self.learning.ml_progress_metrics() return { "service": "hyperliquid-adaptive-bot", "status": "ok" if self._running else "stopping", "timestamp": datetime.now(timezone.utc).isoformat(), "paper_equity": self.paper.equity, "open_positions_count": len(self.paper.positions), "trades": int(metrics["trades"]), "win_rate": metrics["win_rate"], "profit_factor": metrics["profit_factor"], "brier_rolling_50": ml_progress["brier_rolling_50"], "expectancy_usd_30": ml_progress["expectancy_usd_30"], "max_drawdown_pct": ml_progress["max_drawdown_pct"], "feature_drift_score": ml_progress["feature_drift_score"], "edge_rolling_50": ml_progress["edge_rolling_50"], "consistently_profitable": self.learning.consistently_profitable(), "live_armed": self._live_clients is not None, "live_approval_required": self.require_live_approval, "live_approval_granted": self._live_approval_granted(), "live_startup_test_done": self._live_startup_test_done, "live_startup_test_passed": self._live_startup_test_passed, "advisor_enabled": self.hf_advisor.enabled, "advisor_remote_enabled": self.hf_advisor.remote_enabled, "advisor_mode": self.hf_advisor_mode, } def _apply_hf_advisor( self, coin: str, features: List[float], candle: Dict[str, float], now_ts: int, p_long: float, threshold: float, side: Optional[str], ) -> Tuple[Optional[str], str, Optional[float], Optional[str], Optional[float]]: mode = self.hf_advisor_mode self._advisor_stats["requests"] += 1 advice = self.hf_advisor.advise( coin=coin, interval=self.interval, features=features, candle=candle, timestamp_ms=now_ts, ) if advice is None: return side, "advisor_unavailable", p_long, None, None if advice.error: self._advisor_stats["errors"] += 1 if advice.source == "remote": self._advisor_stats["remote_used"] += 1 else: self._advisor_stats["local_used"] += 1 advisor_side = advice.side advisor_conf = float(advice.confidence) if mode == "blend" and advice.p_long is not None: blended_p = (1.0 - self.hf_advisor_blend_weight) * p_long + self.hf_advisor_blend_weight * float(advice.p_long) blended_p = float(np.clip(blended_p, 0.01, 0.99)) blended_side: Optional[str] = None if blended_p >= threshold: blended_side = "long" elif blended_p <= (1.0 - threshold): blended_side = "short" if blended_side is None: self._advisor_stats["held_by_advisor"] += 1 return None, "advisor_hold_blend", blended_p, advisor_side, advisor_conf self._advisor_stats["used_advisor_blend"] += 1 return blended_side, "advisor_blend", blended_p, advisor_side, advisor_conf if side is None: return None, "advisor_no_model_side", p_long, advisor_side, advisor_conf if mode == "confirm": if advisor_side == side and advisor_conf >= self.hf_advisor_min_confidence: self._advisor_stats["kept_model_signal"] += 1 return side, "advisor_confirm", p_long, advisor_side, advisor_conf self._advisor_stats["held_by_advisor"] += 1 return None, "advisor_block_confirm", p_long, advisor_side, advisor_conf if mode == "veto": opposite = (side == "long" and advisor_side == "short") or (side == "short" and advisor_side == "long") if opposite and advisor_conf >= self.hf_advisor_min_confidence: self._advisor_stats["held_by_advisor"] += 1 return None, "advisor_veto", p_long, advisor_side, advisor_conf self._advisor_stats["kept_model_signal"] += 1 return side, "advisor_pass", p_long, advisor_side, advisor_conf return side, "advisor_pass", p_long, advisor_side, advisor_conf def open_trades_payload(self) -> Dict[str, Any]: return { "service": "hyperliquid-adaptive-bot", "status": "ok" if self._running else "stopping", "timestamp": datetime.now(timezone.utc).isoformat(), "open_positions_count": len(self.paper.positions), "open_positions": [asdict(p) for p in self.paper.positions.values()], } def _restore_runtime_state(self) -> None: equity, positions, entry_features = self.learning.restore_runtime_state() if equity is not None: self.paper.equity = float(equity) if positions: self.paper.positions = positions if entry_features: self._entry_features = entry_features if equity is not None or positions: logger.info( "Restored paper runtime state: equity=%.2f open_positions=%d tracked_entries=%d", self.paper.equity, len(self.paper.positions), len(self._entry_features), ) def _persist_runtime_state(self) -> None: self.learning.record_equity(self.paper.equity) self.learning.set_runtime_state( paper_equity=self.paper.equity, positions=self.paper.positions, entry_features=self._entry_features, ) def _entry_pause_reason(self) -> Optional[str]: # Check for 5 consecutive losses — pause immediately without requiring drawdown. recent_pnl = [float(t.get("pnl_usd", 0.0)) for t in self.learning.closed_trades[-5:]] if len(recent_pnl) >= 5 and all(p < 0 for p in recent_pnl): now_t = time.time() if self._consec_loss_pause_ts is None: self._consec_loss_pause_ts = now_t logger.warning("Safety pause: 5 consecutive losses. Pausing entries.") if self.pause_cooldown_minutes > 0: elapsed = now_t - self._consec_loss_pause_ts cooldown = self.pause_cooldown_minutes * 60.0 if elapsed < cooldown: remaining = int(cooldown - elapsed) return f"safety_pause: 5 consecutive losses (cooldown {remaining // 60:02d}m{remaining % 60:02d}s remaining)" else: self._consec_loss_pause_ts = None else: return "safety_pause: 5 consecutive losses" else: self._consec_loss_pause_ts = None if not self.pause_on_negative_expectancy: self._safety_pause_started_ts = None self._safety_pause_wait_recovery = False return None pnl = [float(t.get("pnl_usd", 0.0)) for t in self.learning.closed_trades] if len(pnl) < self.pause_expectancy_lookback: self._safety_pause_started_ts = None self._safety_pause_wait_recovery = False return None exp_recent = float(np.mean(pnl[-self.pause_expectancy_lookback:])) if exp_recent >= self.pause_expectancy_threshold_usd: if self._safety_pause_wait_recovery: logger.info("Safety pause re-armed after expectancy recovered (expectancy_%d=%+.3f).", self.pause_expectancy_lookback, exp_recent) self._safety_pause_started_ts = None self._safety_pause_wait_recovery = False return None max_dd = self.learning.ml_progress_metrics().get("max_drawdown_pct") max_dd_val = float(max_dd) if max_dd is not None else 0.0 if max_dd_val < self.pause_max_drawdown_pct: if self._safety_pause_wait_recovery: logger.info("Safety pause re-armed after drawdown recovered (max_dd=%.2f%%).", max_dd_val) self._safety_pause_started_ts = None self._safety_pause_wait_recovery = False return None if self._safety_pause_wait_recovery: return None reason = ( f"safety_pause expectancy_{self.pause_expectancy_lookback}={exp_recent:+.3f} " f"< {self.pause_expectancy_threshold_usd:+.3f} and max_dd={max_dd_val:.2f}%" ) if self.pause_cooldown_minutes <= 0: return reason now_ts = time.time() if self._safety_pause_started_ts is None: self._safety_pause_started_ts = now_ts cooldown_seconds = self.pause_cooldown_minutes * 60.0 elapsed_seconds = now_ts - self._safety_pause_started_ts if elapsed_seconds >= cooldown_seconds: self._safety_pause_started_ts = None self._safety_pause_wait_recovery = True logger.warning( "Safety pause cooldown expired after %.1f minutes; allowing entries until metrics recover.", self.pause_cooldown_minutes, ) return None remaining_seconds = int(max(0.0, cooldown_seconds - elapsed_seconds)) remaining_minutes = remaining_seconds // 60 remaining_remainder = remaining_seconds % 60 return ( f"{reason} (cooldown {remaining_minutes:02d}m{remaining_remainder:02d}s remaining)" ) def _live_approval_granted(self) -> bool: return self._live_approval_status_details()["granted"] def _live_approval_status_details(self) -> Dict[str, Any]: env_approved = os.environ.get("ALLOW_LIVE_TRADING", "").strip().lower() in {"1", "true", "yes", "y"} file_approved = bool(self.live_approval_file and os.path.exists(self.live_approval_file)) mode = (self.live_approval_mode or "either").strip().lower() missing: List[str] = [] if mode == "both": granted = env_approved and file_approved if not env_approved: missing.append("env:ALLOW_LIVE_TRADING") if not file_approved: missing.append(f"file:{self.live_approval_file}") else: granted = env_approved or file_approved if not granted: missing = ["env:ALLOW_LIVE_TRADING", f"file:{self.live_approval_file}"] return { "mode": mode, "env_approved": env_approved, "file_approved": file_approved, "missing": missing, "granted": granted, } def _maybe_notify_live_readiness(self) -> None: if self._live_readiness_alerted or not self.learning.consistently_profitable(): return m = self.learning.global_metrics() msg = ( "Paper performance gate reached.\n" f"Trades: {int(m['trades'])}, win rate: {m['win_rate'] * 100.0:.1f}%, PF: {m['profit_factor']:.2f}.\n" "Grant live approval to arm live mode. " f"Approval mode={self.live_approval_mode}. " f"Set ALLOW_LIVE_TRADING=1 and/or create {self.live_approval_file} (based on mode)." ) logger.warning(msg) if self.notifications_enabled: self.notifier.send_info(msg) self._live_readiness_alerted = True def _live_order_filled(self, result: Any) -> bool: if not isinstance(result, dict): return False try: statuses = result.get("response", {}).get("data", {}).get("statuses", []) errors = [s.get("error") for s in statuses if "error" in s] if errors: return False filled_sz = sum(float(s["filled"].get("totalSz", 0.0)) for s in statuses if "filled" in s) return filled_sz > 0 except Exception: return False def _calc_test_size(self, coin: str, px: float, min_notional: float) -> float: sz_dec = self.api.get_sz_decimals(coin) step = 10 ** (-sz_dec) raw = min_notional / max(px, 1e-9) steps = max(1, int(math.ceil(raw / step))) return float(round(steps * step, sz_dec)) def _run_live_startup_test(self) -> None: if self._live_clients is None or self._live_startup_test_done or not self.live_startup_test: return self._live_startup_test_done = True coin = self.live_startup_test_coin exchange = self._live_clients["exchange"] try: df = self._fetch_df(coin, lookback_bars=120) if df is None: raise RuntimeError(f"No candle data available for startup live test on {coin}") entry_px = float(df.iloc[-1]["close"]) test_size = self._calc_test_size(coin=coin, px=entry_px, min_notional=self.live_startup_test_notional) logger.warning( "Running startup live order test: BUY then SELL %s size=%.8f around px=%.4f", coin, test_size, entry_px, ) buy_result = self.api.place_market_order( exchange=exchange, coin=coin, is_buy=True, sz=test_size, px=entry_px, sl=None, tp=None, ) time.sleep(1.2) sell_result = self.api.place_market_order( exchange=exchange, coin=coin, is_buy=False, sz=test_size, px=entry_px, sl=None, tp=None, ) buy_ok = self._live_order_filled(buy_result) sell_ok = self._live_order_filled(sell_result) self._live_startup_test_passed = bool(buy_ok and sell_ok) if self._live_startup_test_passed: msg = f"Startup live order test passed on {coin} (open+close filled)." logger.warning(msg) if self.notifications_enabled: self.notifier.send_info(msg) else: msg = f"Startup live order test failed on {coin}. Review logs/order responses before trusting live mirroring." logger.error(msg) if self.notifications_enabled: self.notifier.send_error("adaptive_bot", msg) except Exception as exc: self._live_startup_test_passed = False msg = f"Startup live order test exception: {exc}" logger.error(msg) if self.notifications_enabled: self.notifier.send_error("adaptive_bot", msg) def _start_health_server(self) -> None: bot = self class HealthHandler(BaseHTTPRequestHandler): def _write_json(self, payload: Dict[str, Any], status: int = 200) -> None: body = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _write_head(self, status: int = 200) -> None: self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", "0") self.end_headers() def do_GET(self) -> None: # noqa: N802 if self.path in ("/", "/healthz"): self._write_json({"status": "ok", "service": "hyperliquid-adaptive-bot"}) return if self.path == "/status": self._write_json(bot.status_overview_payload()) return if self.path == "/status/full": self._write_json(bot.status_payload()) return if self.path == "/trades": self._write_json(bot.open_trades_payload()) return self._write_json({"status": "not_found"}, status=404) def do_HEAD(self) -> None: # noqa: N802 if self.path in ("/", "/healthz", "/status", "/status/full", "/trades"): self._write_head(status=200) return self._write_head(status=404) def log_message(self, format: str, *args: Any) -> None: logger.debug("health_http: " + format, *args) server = ThreadingHTTPServer((self.health_host, self.health_port), HealthHandler) self._http_server = server t = threading.Thread(target=server.serve_forever, daemon=True) t.start() logger.info("Health endpoint listening on http://%s:%s", self.health_host, self.health_port) def _fetch_df(self, coin: str, lookback_bars: int = 200) -> Optional[pd.DataFrame]: start = int((datetime.now(timezone.utc) - timedelta(milliseconds=self._interval_ms() * lookback_bars)).timestamp() * 1000) raw = self.api.get_candles(coin=coin, interval=self.interval, start_time=start) if not raw: return None df = parse_candles(raw) if len(df) < 80: return None return df def _size_for_trade(self, entry: float, stop: float, risk_multiplier: float = 1.0) -> float: applied_multiplier = float(np.clip(risk_multiplier, self.risk_volatility_floor_mult, self.risk_volatility_cap_mult)) max_risk_usd = self.paper.equity * self.risk_per_trade * applied_multiplier stop_distance = abs(entry - stop) if stop_distance <= 0: return 0.0 size = max_risk_usd / stop_distance max_notional = self.paper.equity * 0.25 if size * entry > max_notional: size = max_notional / entry return float(max(size, 0.0)) def _risk_profile_for_trade(self, coin: str, features: List[float]) -> Dict[str, Any]: regime = self.learning.classify_regime(features) regime_mult_map = { "low": 1.05, "normal": 1.0, "high": 0.75, "extreme": 0.55, } volatility_mult = float(regime_mult_map.get(regime, 1.0)) perf = self.learning.regime_performance(coin, regime) perf_mult = 1.0 if ( perf.get("trades") is not None and int(perf["trades"]) >= self.risk_regime_min_samples and perf.get("win_rate") is not None ): edge = (float(perf["win_rate"]) - 0.5) * 2.0 perf_mult = 1.0 + self.risk_regime_adapt_strength * 0.35 * edge combined = float(np.clip(volatility_mult * perf_mult, self.risk_volatility_floor_mult, self.risk_volatility_cap_mult)) return { "regime": regime, "volatility_mult": volatility_mult, "performance_mult": perf_mult, "risk_multiplier": combined, "regime_stats": perf, } def _maybe_prepare_live(self) -> None: if not self.enable_live: return if not self.learning.consistently_profitable(): return if self.require_live_approval and not self._live_approval_granted(): return if self._live_clients is not None: return pk = os.environ.get("HL_PRIVATE_KEY", "").strip() acct = os.environ.get("HL_ACCOUNT_ADDRESS", "").strip() or None if not pk: logger.warning("Live mode requested and paper criteria met, but HL_PRIVATE_KEY is missing.") return try: self._live_clients = self.api.create_live_execution_clients(private_key=pk, account_address=acct) logger.warning("Paper criteria met. Live execution is now armed.") self._run_live_startup_test() except Exception as exc: logger.error("Failed to initialize live execution clients: %s", exc) self._live_clients = None def _place_live_mirror(self, coin: str, side: str, entry: float, sl: float, tp: float, size: float) -> None: if self._live_clients is None: return exchange = self._live_clients["exchange"] is_buy = side == "long" try: self.api.place_market_order(exchange=exchange, coin=coin, is_buy=is_buy, sz=size, px=entry, sl=sl, tp=tp) except Exception as exc: logger.error("Live mirror order failed for %s: %s", coin, exc) def _notify_trade_open(self, coin: str, side: str, entry: float, sl: float, tp: float, size: float) -> None: if not self.notifications_enabled: return rr: Optional[float] = None risk = abs(entry - sl) reward = abs(tp - entry) if risk > 0: rr = reward / risk self.notifier.send_trade( direction=side.upper(), coin=coin, entry=entry, sl=sl, tp=tp, sz=size, notional=entry * size, strategy="adaptive_ml", timeframe=self.interval, rr=rr, order_type="paper", ) def _notify_trade_close(self, closed: ClosedTrade) -> None: if not self.notifications_enabled: return self.notifier.send_exit( coin=closed.coin, reason=closed.reason, pnl_pct=closed.pnl_pct, pnl_usdc=closed.pnl_usd, strategy="adaptive_ml", timeframe=self.interval, ) def _maybe_send_daily_summary(self) -> None: if not self.notifications_enabled: return now = datetime.now(timezone.utc) date_key = now.date().isoformat() if now.hour != 0 or self._last_summary_date == date_key: return since_ms = int((now - timedelta(days=1)).timestamp() * 1000) day_trades = [ t for t in self.learning.closed_trades if int(t.get("exit_ts", 0)) >= since_ms ] day_wins = sum(1 for t in day_trades if float(t.get("pnl_usd", 0.0)) > 0) day_wr = (day_wins / len(day_trades) * 100.0) if day_trades else 0.0 day_pnl = sum(float(t.get("pnl_usd", 0.0)) for t in day_trades) overall = self.learning.global_metrics() ml = self.learning.ml_progress_metrics() summary = ( "Midnight summary (UTC)\n" f"24h trades: {len(day_trades)}\n" f"24h win rate: {day_wr:.1f}%\n" f"24h pnl: {day_pnl:+.2f} USDC\n" f"Total trades: {int(overall['trades'])}\n" f"Total win rate: {overall['win_rate'] * 100.0:.1f}%\n" f"Profit factor: {overall['profit_factor']:.2f}\n" f"Paper equity: {self.paper.equity:.2f}\n" f"Brier(50): {ml['brier_rolling_50']}\n" f"Expectancy(30): {ml['expectancy_usd_30']}\n" f"Max DD: {ml['max_drawdown_pct']}%" ) self.notifier.send_info(summary) self._last_summary_date = date_key def _maybe_send_ml_progress_report(self) -> None: if not self.notifications_enabled: return if self.telegram_ml_report_hours <= 0: return now_ts = time.time() period_s = self.telegram_ml_report_hours * 3600.0 if self._last_ml_report_ts > 0 and (now_ts - self._last_ml_report_ts) < period_s: return overall = self.learning.global_metrics() trades = int(overall["trades"]) if trades < self.telegram_ml_report_min_trades: return ml = self.learning.ml_progress_metrics() per_coin = ml.get("per_coin", {}) coin_parts = [] for coin in ALLOWED_COINS: coin_brier = per_coin.get(coin, {}).get("brier_rolling_50") coin_parts.append(f"{coin}:{coin_brier}") coin_brier_line = " | ".join(coin_parts) msg = ( "ML progress update\n" f"Trades: {trades}\n" f"Win rate: {overall['win_rate'] * 100.0:.1f}%\n" f"Profit factor: {overall['profit_factor']:.2f}\n" f"Brier(50): {ml['brier_rolling_50']}\n" f"Expectancy(30): {ml['expectancy_usd_30']}\n" f"Max DD: {ml['max_drawdown_pct']}%\n" f"Per-coin Brier(50): {coin_brier_line}" ) self.notifier.send_info(msg) self._last_ml_report_ts = now_ts def _record_closed_trade(self, closed: ClosedTrade, coin: str) -> None: """Shared post-close logic: notify, learn, persist.""" logger.info( "CLOSED %s %s pnl=$%.2f (%+.2f%%) reason=%s", closed.coin, closed.side, closed.pnl_usd, closed.pnl_pct, closed.reason, ) self._notify_trade_close(closed) self.learning.record_trade_event("close", asdict(closed)) stored = self._entry_features.get((coin, closed.entry_ts), []) if stored: is_bootstrap = (coin, closed.entry_ts) in self._bootstrap_entries self._bootstrap_entries.discard((coin, closed.entry_ts)) self.learning.update_after_trade(closed, stored, skip_model_fit=is_bootstrap) self._entry_features.pop((coin, closed.entry_ts), None) self._persist_runtime_state() def _trade_coin(self, coin: str) -> None: df = self._fetch_df(coin) if df is None: logger.warning("Not enough candle data for %s", coin) return candle = { "high": float(df.iloc[-1]["high"]), "low": float(df.iloc[-1]["low"]), "close": float(df.iloc[-1]["close"]), } now_ts = int(df.iloc[-1]["ts"]) # Fetch perp market context and HTF trend for feature construction. funding_rate, oi_change_pct = self._fetch_market_context(coin) htf_trend = self._compute_htf_trend(df) # SL/TP check first. closed = self.paper.maybe_close(coin, candle, now_ts) if closed: self._record_closed_trade(closed, coin) # If still in a position, check for a model-driven early exit. if self.paper.has_position(coin): if self.learning.model_ready.get(coin): features = build_features(df, funding_rate, oi_change_pct, htf_trend) if features: p_long = self.learning.predict_long_probability(coin, features) threshold = self.learning.thresholds[coin] pos = self.paper.positions[coin] should_exit = ( pos.side == "long" and p_long < (1.0 - threshold - 0.05) ) or ( pos.side == "short" and p_long > (threshold + 0.05) ) if should_exit: early_closed = self.paper.force_close(coin, candle["close"], now_ts, "model_exit") if early_closed: logger.info( "MODEL EXIT %s %s pnl=$%.2f p_long=%.3f reversed past threshold", coin, early_closed.side, early_closed.pnl_usd, p_long, ) self._record_closed_trade(early_closed, coin) return # don't enter new position this cycle features = build_features(df, funding_rate, oi_change_pct, htf_trend) if not features: return # Virtual held-bar training: previous cycle's features + this bar's direction = free label. # Weight is low (_VIRTUAL_SAMPLE_WEIGHT) so real trade labels dominate. prev_feats = self._prev_bar_features.get(coin) prev_close = self._prev_bar_close.get(coin) curr_close = candle["close"] if prev_feats is not None and prev_close is not None and prev_close > 0: bar_label = 1 if curr_close > prev_close else 0 self.learning.train_virtual_sample(coin, prev_feats, bar_label) self._prev_bar_features[coin] = features self._prev_bar_close[coin] = curr_close pause_reason = self._entry_pause_reason() if pause_reason: logger.warning("%s HOLD %s", coin, pause_reason) return p_long = self.learning.predict_long_probability(coin, features) threshold = self.learning.thresholds[coin] side: Optional[str] = None signal_reason = "model" if not self.learning.enough_samples(coin, self.min_model_samples_for_signal): # Bootstrap: trade weak momentum to generate first training labels. # These trades are excluded from model.partial_fit to avoid polluting early weights. side = "long" if features[1] >= 0 else "short" signal_reason = "bootstrap" else: if p_long >= threshold: side = "long" elif p_long <= (1.0 - threshold): side = "short" if side is None and random.random() < self.exploration_rate: side = "long" if p_long >= 0.5 else "short" signal_reason = "exploration" side, signal_reason, p_long, advisor_side, advisor_conf = self._apply_hf_advisor( coin=coin, features=features, candle=candle, now_ts=now_ts, p_long=p_long, threshold=threshold, side=side, ) if not side: _now = time.time() if _now - self._last_hold_log_ts.get(coin, 0.0) >= 600: if advisor_side is not None: logger.info( "%s HOLD p_long=%.3f threshold=%.3f advisor=%s conf=%.3f reason=%s", coin, p_long, threshold, advisor_side, advisor_conf if advisor_conf is not None else 0.0, signal_reason, ) else: logger.info("%s HOLD p_long=%.3f threshold=%.3f", coin, p_long, threshold) self._last_hold_log_ts[coin] = _now return entry = candle["close"] atr = atr_for_risk(df) # Dynamic TP: scale reward target with model confidence so high-conviction # trades are given more room to run (SL stays fixed to protect capital). confidence = float(np.clip(abs(p_long - 0.5) * 2.0, 0.0, 1.0)) tp_mult_dynamic = self.tp_mult * (1.0 + 0.4 * confidence) if side == "long": sl = max(entry - atr * self.sl_mult, 1e-8) tp = entry + atr * tp_mult_dynamic else: sl = entry + atr * self.sl_mult tp = max(entry - atr * tp_mult_dynamic, 1e-8) risk_profile = self._risk_profile_for_trade(coin, features) size = self._size_for_trade(entry, sl, risk_profile["risk_multiplier"]) if size <= 0: return self._last_risk_snapshot = { "coin": coin, "ts": now_ts, "regime": risk_profile["regime"], "volatility_mult": risk_profile["volatility_mult"], "performance_mult": risk_profile["performance_mult"], "risk_multiplier": risk_profile["risk_multiplier"], "atr": atr, "base_risk_per_trade": self.risk_per_trade, "effective_risk_per_trade": self.risk_per_trade * risk_profile["risk_multiplier"], "regime_stats": risk_profile["regime_stats"], "tp_mult_dynamic": tp_mult_dynamic, "confidence": confidence, } pos = Position( coin=coin, side=side, entry=entry, size=size, sl=sl, tp=tp, entry_ts=now_ts, entry_features=features, entry_p_long=p_long, signal_reason=signal_reason, ) self.paper.open_position(pos) self._entry_features[(coin, now_ts)] = features if signal_reason == "bootstrap": self._bootstrap_entries.add((coin, now_ts)) self.learning.record_trade_event( "open", { "coin": coin, "side": side, "entry": entry, "size": size, "sl": sl, "tp": tp, "entry_ts": now_ts, "signal_reason": signal_reason, "p_long": p_long, "threshold": threshold, "advisor_side": advisor_side, "advisor_confidence": advisor_conf, "risk_regime": risk_profile["regime"], "risk_multiplier": risk_profile["risk_multiplier"], "risk_volatility_mult": risk_profile["volatility_mult"], "risk_performance_mult": risk_profile["performance_mult"], "tp_mult_dynamic": tp_mult_dynamic, "funding_rate": funding_rate, "oi_change_pct": oi_change_pct, "htf_trend": htf_trend, }, ) self._persist_runtime_state() logger.info( "OPEN %s %s entry=%.4f sl=%.4f tp=%.4f size=%.5f p_long=%.3f th=%.3f " "tp_mult=%.2f regime=%s risk_mult=%.3f reason=%s equity=%.2f", coin, side, entry, sl, tp, size, p_long, threshold, tp_mult_dynamic, risk_profile["regime"], risk_profile["risk_multiplier"], signal_reason, self.paper.equity, ) self._notify_trade_open(coin, side, entry, sl, tp, size) self._maybe_prepare_live() self._place_live_mirror(coin, side, entry, sl, tp, size) def write_report(self) -> None: payload = self.status_payload() with open(self.report_file, "w", encoding="utf-8") as f: json.dump(payload, f, indent=2) def run(self) -> None: signal.signal(signal.SIGINT, self.stop) signal.signal(signal.SIGTERM, self.stop) self._start_health_server() logger.info("Starting adaptive bot (paper mode=%s, live enabled=%s)", True, self.enable_live) logger.info("Universe: %s", ", ".join(ALLOWED_COINS)) while self._running: try: for coin in ALLOWED_COINS: self._trade_coin(coin) self._persist_runtime_state() self.learning.save() self.write_report() now_ts = time.time() m = self.learning.global_metrics() if now_ts - self._last_stats_log_ts >= 600: logger.info( "Stats trades=%d win_rate=%.2f%% pf=%.2f paper_equity=%.2f profitable_gate=%s", int(m["trades"]), m["win_rate"] * 100.0, m["profit_factor"], self.paper.equity, self.learning.consistently_profitable(), ) self._last_stats_log_ts = now_ts self._maybe_notify_live_readiness() self._maybe_prepare_live() self._maybe_send_daily_summary() self._maybe_send_ml_progress_report() except Exception as exc: logger.exception("Loop error: %s", exc) time.sleep(self.poll_seconds) def build_arg_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser(description="Adaptive ML Hyperliquid paper trader") p.add_argument("--interval", default="15m", choices=sorted(INTERVAL_MS.keys())) p.add_argument("--state-file", default="learning_state.joblib") p.add_argument("--report-file", default="paper_report.json") p.add_argument("--starting-equity", type=float, default=10000.0) p.add_argument("--risk-per-trade", type=float, default=0.01, help="Fraction of equity risked per position") p.add_argument("--poll-seconds", type=int, default=45) p.add_argument("--exploration-rate", type=float, default=0.08, help="Chance to take an exploratory trade when model signal is neutral") p.add_argument("--health-host", default=os.environ.get("BOT_HEALTH_HOST", "0.0.0.0")) p.add_argument("--health-port", type=int, default=int(os.environ.get("PORT", os.environ.get("BOT_HEALTH_PORT", "7860")))) p.add_argument("--enable-live", action="store_true", help="Allow live mirror orders after paper criteria is met") p.add_argument( "--require-live-approval", action=argparse.BooleanOptionalAction, default=True, help="Require manual approval before arming live mirror orders", ) p.add_argument( "--live-approval-mode", choices=["either", "both"], default="either", help="Manual approval policy: 'either' (env or file) or 'both' (env and file)", ) p.add_argument("--live-approval-file", default="allow_live_trading.flag", help="Approval file path to authorize live trading") p.add_argument( "--live-startup-test", action=argparse.BooleanOptionalAction, default=False, help="When live is armed, place a tiny BUY then SELL to verify live order execution", ) p.add_argument("--live-startup-test-coin", choices=sorted(ALLOWED_COINS), default=ALLOWED_COINS[0]) p.add_argument( "--live-startup-test-notional", type=float, default=12.0, help="Approx USDC notional used for startup live order test (min effectively 10)", ) p.add_argument("--hf-advisor-url", default=os.environ.get("BOT_HF_ADVISOR_URL", ""), help="Optional external advisor endpoint URL") p.add_argument("--hf-advisor-token", default=os.environ.get("BOT_HF_ADVISOR_TOKEN", ""), help="Optional bearer token for advisor endpoint") p.add_argument( "--hf-advisor-mode", choices=["confirm", "veto", "blend"], default=os.environ.get("BOT_HF_ADVISOR_MODE", "blend"), help="confirm=must match advisor, veto=advisor can block, blend=blend p_long with advisor p_long", ) p.add_argument( "--hf-advisor-min-confidence", type=float, default=float(os.environ.get("BOT_HF_ADVISOR_MIN_CONFIDENCE", "0.60")), help="Minimum advisor confidence used by confirm/veto modes", ) p.add_argument( "--hf-advisor-blend-weight", type=float, default=float(os.environ.get("BOT_HF_ADVISOR_BLEND_WEIGHT", "0.0")), help="Advisor p_long blending weight for blend mode [0..1] (0 = advisor disabled unless remote endpoint set)", ) p.add_argument( "--hf-advisor-timeout-seconds", type=float, default=float(os.environ.get("BOT_HF_ADVISOR_TIMEOUT_SECONDS", "4.0")), help="HTTP timeout for advisor requests", ) p.add_argument( "--min-model-samples-for-signal", type=int, default=int(os.environ.get("BOT_MIN_MODEL_SAMPLES_FOR_SIGNAL", "50")), help="Minimum labeled samples per coin before model-driven entries are trusted", ) p.add_argument( "--pause-on-negative-expectancy", action=argparse.BooleanOptionalAction, default=(os.environ.get("BOT_PAUSE_ON_NEGATIVE_EXPECTANCY", "1").strip().lower() in {"1", "true", "yes", "y"}), help="Pause new entries when recent expectancy is poor and drawdown is elevated", ) p.add_argument( "--pause-expectancy-lookback", type=int, default=int(os.environ.get("BOT_PAUSE_EXPECTANCY_LOOKBACK", "30")), help="Number of latest closed trades used for safety pause expectancy", ) p.add_argument( "--pause-expectancy-threshold-usd", type=float, default=float(os.environ.get("BOT_PAUSE_EXPECTANCY_THRESHOLD_USD", "-1.0")), help="Pause if recent expectancy is below this USD per-trade threshold", ) p.add_argument( "--pause-max-drawdown-pct", type=float, default=float(os.environ.get("BOT_PAUSE_MAX_DRAWDOWN_PCT", "12.0")), help="Pause only when max drawdown is also at/above this percentage", ) p.add_argument( "--pause-cooldown-minutes", type=float, default=float(os.environ.get("BOT_PAUSE_COOLDOWN_MINUTES", "180")), help="Minutes to keep safety pause active before auto-resuming entries until metrics recover (0 disables auto-expiry)", ) p.add_argument( "--telegram-ml-report-hours", type=float, default=float(os.environ.get("BOT_TELEGRAM_ML_REPORT_HOURS", "6")), help="Send periodic ML progress updates to Telegram every N hours (0 disables)", ) p.add_argument( "--telegram-ml-report-min-trades", type=int, default=int(os.environ.get("BOT_TELEGRAM_ML_REPORT_MIN_TRADES", "15")), help="Minimum number of closed trades before periodic ML Telegram reports are sent", ) p.add_argument( "--sl-mult", type=float, default=float(os.environ.get("BOT_SL_MULT", "1.3")), help="Stop-loss distance as multiple of ATR14 (default 1.3)", ) p.add_argument( "--tp-mult", type=float, default=float(os.environ.get("BOT_TP_MULT", "2.0")), help="Take-profit distance as multiple of ATR14 (default 2.0, giving ~1:1.5 R:R)", ) p.add_argument( "--risk-volatility-floor-mult", type=float, default=float(os.environ.get("BOT_RISK_VOLATILITY_FLOOR_MULT", "0.45")), help="Minimum risk multiplier after volatility/regime adjustment", ) p.add_argument( "--risk-volatility-cap-mult", type=float, default=float(os.environ.get("BOT_RISK_VOLATILITY_CAP_MULT", "1.20")), help="Maximum risk multiplier after volatility/regime adjustment", ) p.add_argument( "--risk-regime-adapt-strength", type=float, default=float(os.environ.get("BOT_RISK_REGIME_ADAPT_STRENGTH", "0.35")), help="How strongly recent regime performance adjusts risk sizing [0..1]", ) p.add_argument( "--risk-regime-min-samples", type=int, default=int(os.environ.get("BOT_RISK_REGIME_MIN_SAMPLES", "12")), help="Minimum closed trades in a regime before performance-based risk adjustment applies", ) return p def main() -> None: args = build_arg_parser().parse_args() bot = AdaptiveHyperliquidBot( interval=args.interval, state_file=args.state_file, report_file=args.report_file, starting_equity=args.starting_equity, risk_per_trade=args.risk_per_trade, poll_seconds=args.poll_seconds, exploration_rate=args.exploration_rate, enable_live=args.enable_live, require_live_approval=args.require_live_approval, live_approval_mode=args.live_approval_mode, live_approval_file=args.live_approval_file, live_startup_test=args.live_startup_test, live_startup_test_coin=args.live_startup_test_coin, live_startup_test_notional=args.live_startup_test_notional, hf_advisor_url=args.hf_advisor_url, hf_advisor_token=args.hf_advisor_token, hf_advisor_mode=args.hf_advisor_mode, hf_advisor_min_confidence=args.hf_advisor_min_confidence, hf_advisor_blend_weight=args.hf_advisor_blend_weight, hf_advisor_timeout_seconds=args.hf_advisor_timeout_seconds, min_model_samples_for_signal=args.min_model_samples_for_signal, pause_on_negative_expectancy=args.pause_on_negative_expectancy, pause_expectancy_lookback=args.pause_expectancy_lookback, pause_expectancy_threshold_usd=args.pause_expectancy_threshold_usd, pause_max_drawdown_pct=args.pause_max_drawdown_pct, pause_cooldown_minutes=args.pause_cooldown_minutes, telegram_ml_report_hours=args.telegram_ml_report_hours, telegram_ml_report_min_trades=args.telegram_ml_report_min_trades, health_host=args.health_host, health_port=args.health_port, sl_mult=args.sl_mult, tp_mult=args.tp_mult, risk_volatility_floor_mult=args.risk_volatility_floor_mult, risk_volatility_cap_mult=args.risk_volatility_cap_mult, risk_regime_adapt_strength=args.risk_regime_adapt_strength, risk_regime_min_samples=args.risk_regime_min_samples, ) bot.run() if __name__ == "__main__": main()