""" RSI Mean Reversion Bot v2 — Self-Optimizing ============================================= Trades the TOP 3 coins on Kraken by RSI signal quality. Every night at midnight, scans 20 candidate coins and automatically replaces underperformers with better opportunities. Strategy: BUY when RSI < 30 (oversold) + price above MA (trend filter) SELL when RSI > 70 (overbought) OR +4% take profit OR -2% stop loss Self-optimization: - Scores all 20 candidates daily on volatility, trend, volume, signal freq - Swaps out any coin scoring 20+ points below the best available alternative - Never swaps a coin with an active open position - Notifies you on Mac + logs every swap decision with full reasoning Capital: $275 split evenly across 3 active coins (~$91.67 each) Expected: $50-100/month at moderate volatility Setup: pip install krakenex Add API keys to config_advanced.py or .env Run: python3 rsi_bot_v2.py """ import krakenex import time import json import logging import os from datetime import datetime, timedelta from collections import deque # ── API Keys ────────────────────────────────────────────────────────────────── try: from config_advanced import API_KEY, API_SECRET except ImportError: try: from dotenv import load_dotenv load_dotenv() except ImportError: try: with open('.env') as f: for line in f: if '=' in line: k, v = line.strip().split('=', 1) os.environ[k] = v except FileNotFoundError: pass API_KEY = os.getenv("KRAKEN_API_KEY", "") API_SECRET = os.getenv("KRAKEN_API_SECRET", "") # ── Settings ────────────────────────────────────────────────────────────────── DRY_RUN = False # Set False to go live TOTAL_CAPITAL = 290.0 # Total USD to deploy CHECK_INTERVAL = 60 # Seconds between trading checks OPTIMIZE_HOUR = 0 # Hour to run daily optimization (0 = midnight) RSI_PERIOD = 14 RSI_OVERSOLD = 30 RSI_OVERBOUGHT = 70 TAKE_PROFIT_PCT = 0.04 # 4% STOP_LOSS_PCT = 0.02 # 2% MA_PERIOD = 20 CANDLE_INTERVAL = 60 # 1-hour candles SWAP_SCORE_THRESHOLD = 20 # Min score advantage to trigger a swap MAX_ACTIVE_COINS = 3 # Always trade exactly this many coins # ── Candidate Pool ──────────────────────────────────────────────────────────── # All liquid coins available on Kraken with USD pairs # Bot picks the best 3 from this list every night CANDIDATE_POOL = { "BTC": {"pair": "XXBTZUSD", "min_order": 0.0001, "decimals": 1}, "ETH": {"pair": "XETHZUSD", "min_order": 0.01, "decimals": 2}, "SOL": {"pair": "SOLUSD", "min_order": 0.02, "decimals": 2}, "LINK": {"pair": "LINKUSD", "min_order": 0.1, "decimals": 2}, "DOT": {"pair": "DOTUSD", "min_order": 1.0, "decimals": 3}, "AVAX": {"pair": "AVAXUSD", "min_order": 0.1, "decimals": 2}, "ATOM": {"pair": "ATOMUSD", "min_order": 0.1, "decimals": 3}, "NEAR": {"pair": "NEARUSD", "min_order": 1.0, "decimals": 4}, "UNI": {"pair": "UNIUSD", "min_order": 0.1, "decimals": 3}, "AAVE": {"pair": "AAVEUSD", "min_order": 0.01, "decimals": 2}, "LTC": {"pair": "XLTCZUSD", "min_order": 0.02, "decimals": 2}, "ADA": {"pair": "ADAUSD", "min_order": 10.0, "decimals": 4}, "XRP": {"pair": "XXRPZUSD", "min_order": 10.0, "decimals": 4}, "DOGE": {"pair": "XDGUSD", "min_order": 50.0, "decimals": 5}, "TRX": {"pair": "TRXUSD", "min_order": 100.0, "decimals": 5}, "FIL": {"pair": "FILUSD", "min_order": 0.1, "decimals": 3}, "ALGO": {"pair": "ALGOUSD", "min_order": 5.0, "decimals": 4}, "MANA": {"pair": "MANAUSD", "min_order": 5.0, "decimals": 4}, "SAND": {"pair": "SANDUSD", "min_order": 5.0, "decimals": 4}, "INJ": {"pair": "INJUSD", "min_order": 0.1, "decimals": 3}, } # Starting coins — optimizer will adjust from here ACTIVE_COINS = ["LINK", "DOT", "AVAX"] # ── Logging ─────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler("rsi_bot.log"), logging.StreamHandler(), ] ) log = logging.getLogger(__name__) api = krakenex.API(key=API_KEY, secret=API_SECRET) # ── State ───────────────────────────────────────────────────────────────────── positions = {} # Populated dynamically as coins are activated trade_log = [] total_profit = 0.0 swap_log = [] last_optimize_date = None CAPITAL_PER_COIN = TOTAL_CAPITAL / MAX_ACTIVE_COINS # $91.67 def init_position(name): return { "active": False, "buy_price": 0.0, "volume": 0.0, "order_id": None, "buy_time": None, } for name in ACTIVE_COINS: positions[name] = init_position(name) # ── Notifications ───────────────────────────────────────────────────────────── def notify(title, message=""): try: msg = str(message).replace("'", "").replace('"', '')[:200] t = str(title).replace("'", "").replace('"', '')[:60] os.system(f"osascript -e 'display notification \"{msg}\" with title \"RSI Bot: {t}\"'") except Exception: pass # ── Kraken helpers ──────────────────────────────────────────────────────────── def get_price(pair: str) -> float: try: resp = api.query_public("Ticker", {"pair": pair}) if resp.get("error"): return 0.0 result = resp["result"] key = list(result.keys())[0] return float(result[key]["c"][0]) except Exception: return 0.0 def get_ohlc(pair: str, interval: int = 60, count: int = 250): try: resp = api.query_public("OHLC", {"pair": pair, "interval": interval}) if resp.get("error"): return [] result = resp["result"] key = [k for k in result if k != "last"][0] return result[key][-count:] except Exception: return [] def get_volume_24h(pair: str) -> float: """Get 24h trading volume in USD.""" try: resp = api.query_public("Ticker", {"pair": pair}) if resp.get("error"): return 0.0 result = resp["result"] key = list(result.keys())[0] vol = float(result[key]["v"][1]) # 24h volume in base currency price = float(result[key]["c"][0]) return vol * price except Exception: return 0.0 def place_order(order_type: str, pair: str, volume: float, price: float = None) -> str: if DRY_RUN: oid = f"DRY-{order_type.upper()}-{pair}-{datetime.now().strftime('%H%M%S')}" log.info(f"[DRY RUN] {order_type.upper()} {volume:.4f} {pair} @ ${price or 'market'}") return oid try: params = { "pair": pair, "type": order_type, "ordertype": "limit" if price else "market", "volume": str(round(volume, 6)), } if price: params["price"] = str(price) resp = api.query_private("AddOrder", params) if resp.get("error"): log.error(f"AddOrder error: {resp['error']}") return "" return resp["result"]["txid"][0] except Exception as e: log.error(f"place_order failed: {e}") return "" def cancel_order(order_id: str): if DRY_RUN or not order_id: return try: api.query_private("CancelOrder", {"txid": order_id}) except Exception as e: log.error(f"cancel_order failed: {e}") # ── Indicators ──────────────────────────────────────────────────────────────── def calc_rsi(closes: list, period: int = 14) -> float: if len(closes) < period + 1: return 50.0 gains, losses = [], [] for i in range(1, len(closes)): delta = closes[i] - closes[i - 1] gains.append(max(delta, 0)) losses.append(max(-delta, 0)) avg_gain = sum(gains[-period:]) / period avg_loss = sum(losses[-period:]) / period if avg_loss == 0: return 100.0 rs = avg_gain / avg_loss return round(100 - (100 / (1 + rs)), 2) def calc_ma(closes: list, period: int = 20) -> float: if len(closes) < period: return closes[-1] if closes else 0.0 return sum(closes[-period:]) / period def calc_atr(candles: list, period: int = 14) -> float: if len(candles) < period + 1: return 0.0 trs = [] for i in range(1, len(candles)): high = float(candles[i][2]) low = float(candles[i][3]) close = float(candles[i - 1][4]) trs.append(max(high - low, abs(high - close), abs(low - close))) return sum(trs[-period:]) / period def count_rsi_signals(closes: list, period: int = 14, lookback: int = 30) -> int: """Count how many times RSI crossed oversold/overbought in last N candles.""" if len(closes) < period + lookback: return 0 signals = 0 recent = closes[-(lookback + period):] for i in range(period, len(recent)): rsi = calc_rsi(recent[:i + 1], period) if rsi <= RSI_OVERSOLD or rsi >= RSI_OVERBOUGHT: signals += 1 return signals # ── Coin Scoring Engine ─────────────────────────────────────────────────────── def score_coin(name: str, config: dict) -> dict: """ Score a coin 0-100 for RSI bot suitability. Higher = better trading opportunity right now. Breakdown: 30pts Volatility — ATR% sweet spot 3-8% daily 25pts Trend health — Not in freefall, above 30-day low 20pts Volume — $10M+ daily for reliable fills 15pts Signal freq — RSI hits extremes regularly 10pts Bounce — Recent recovery off lows (momentum reset) """ score = 0 reasons = [] warnings = [] candles = get_ohlc(config["pair"], interval=60, count=200) if len(candles) < 50: return {"name": name, "score": 0, "reasons": ["insufficient data"], "warnings": []} closes = [float(c[4]) for c in candles] current = closes[-1] atr = calc_atr(candles, 14) atr_pct = (atr / current) * 100 if current > 0 else 0 ma20 = calc_ma(closes, 20) low_30d = min(closes[-720:]) if len(closes) >= 720 else min(closes) # 30 days of 1h candles high_30d = max(closes[-720:]) if len(closes) >= 720 else max(closes) vol_24h = get_volume_24h(config["pair"]) rsi_now = calc_rsi(closes) signals = count_rsi_signals(closes, RSI_PERIOD, lookback=30) # ── 1. Volatility score (30 pts) ────────────────────────────────────────── # Sweet spot: 3-8% ATR. Below = dead, above = too wild if atr_pct >= 3.0 and atr_pct <= 8.0: vol_score = 30 reasons.append(f"ideal volatility ATR={atr_pct:.1f}%") elif atr_pct >= 1.5 and atr_pct < 3.0: vol_score = 18 reasons.append(f"moderate volatility ATR={atr_pct:.1f}%") elif atr_pct >= 8.0 and atr_pct <= 15.0: vol_score = 20 reasons.append(f"high volatility ATR={atr_pct:.1f}% (risky but ok)") elif atr_pct > 15.0: vol_score = 8 warnings.append(f"extreme volatility ATR={atr_pct:.1f}% — stop loss may not protect") else: vol_score = 0 warnings.append(f"too quiet ATR={atr_pct:.1f}% — RSI signals rare") score += vol_score # ── 2. Trend health (25 pts) ────────────────────────────────────────────── # We want oscillation, not a freefall. Check: # - Price within 40% of 30d high (not collapsed) # - Price above 30d low by at least 5% (has bounced) pct_from_high = (current - high_30d) / high_30d * 100 # negative = below high pct_from_low = (current - low_30d) / low_30d * 100 # positive = above low if pct_from_high >= -30 and pct_from_low >= 5: trend_score = 25 reasons.append(f"healthy trend — {pct_from_low:.0f}% above 30d low") elif pct_from_high >= -50 and pct_from_low >= 2: trend_score = 15 reasons.append(f"recovering — {pct_from_low:.0f}% off lows") elif pct_from_low < 2: trend_score = 5 warnings.append(f"near 30d low — potential freefall") else: trend_score = 10 warnings.append(f"down {abs(pct_from_high):.0f}% from 30d high — weak trend") score += trend_score # ── 3. Volume (20 pts) ──────────────────────────────────────────────────── if vol_24h >= 50_000_000: score += 20 reasons.append(f"excellent volume ${vol_24h/1e6:.0f}M") elif vol_24h >= 10_000_000: score += 14 reasons.append(f"good volume ${vol_24h/1e6:.0f}M") elif vol_24h >= 3_000_000: score += 7 warnings.append(f"thin volume ${vol_24h/1e6:.1f}M — slippage risk") else: score += 0 warnings.append(f"very low volume ${vol_24h/1e6:.2f}M — skip") # ── 4. RSI signal frequency (15 pts) ───────────────────────────────────── if signals >= 6: score += 15 reasons.append(f"high signal freq ({signals} RSI extremes/30d)") elif signals >= 3: score += 10 reasons.append(f"moderate signal freq ({signals} RSI extremes/30d)") elif signals >= 1: score += 5 reasons.append(f"low signal freq ({signals} RSI extremes/30d)") else: score += 0 warnings.append("no RSI signals in 30d — price too stable") # ── 5. Recent bounce (10 pts) ───────────────────────────────────────────── # Has price recovered from a recent dip? Good for mean reversion recent_low_7d = min(closes[-168:]) if len(closes) >= 168 else min(closes) pct_bounce = (current - recent_low_7d) / recent_low_7d * 100 if pct_bounce >= 5: score += 10 reasons.append(f"strong bounce {pct_bounce:.0f}% off 7d low") elif pct_bounce >= 2: score += 6 reasons.append(f"mild bounce {pct_bounce:.0f}% off 7d low") else: score += 2 warnings.append("no recent bounce — flat or still falling") return { "name": name, "score": score, "atr_pct": round(atr_pct, 2), "vol_24h_m": round(vol_24h / 1e6, 1), "rsi_now": rsi_now, "signals": signals, "reasons": reasons, "warnings": warnings, "current": current, } # ── Daily Optimizer ─────────────────────────────────────────────────────────── def run_daily_optimization(): """ Score all candidate coins. Swap out any active coin that is beaten by 20+ points by a coin not currently active. Never swaps a coin with an open position. """ global ACTIVE_COINS, positions, last_optimize_date log.info("\n" + "=" * 60) log.info("DAILY OPTIMIZATION — Scanning all candidates...") log.info("=" * 60) scores = {} for name, config in CANDIDATE_POOL.items(): try: result = score_coin(name, config) scores[name] = result status = "ACTIVE" if name in ACTIVE_COINS else " " log.info(f" [{status}] {name:6} score={result['score']:3d} | " f"ATR={result['atr_pct']:.1f}% | " f"vol=${result['vol_24h_m']:.0f}M | " f"RSI={result['rsi_now']:.0f} | " f"signals={result['signals']}") time.sleep(0.5) # Avoid rate limiting except Exception as e: log.error(f" Error scoring {name}: {e}") scores[name] = {"name": name, "score": 0, "reasons": [], "warnings": [str(e)]} # Sort all candidates by score ranked = sorted(scores.values(), key=lambda x: x["score"], reverse=True) log.info("\n Top 5 candidates:") for r in ranked[:5]: log.info(f" {r['name']:6} {r['score']:3d}pts — {', '.join(r['reasons'][:2])}") # Determine which active coins should be swapped out swaps_made = [] new_active = list(ACTIVE_COINS) for active_name in list(ACTIVE_COINS): # Never swap if there's an open position if positions.get(active_name, {}).get("active", False): log.info(f" [{active_name}] Skipping — has open position") continue active_score = scores.get(active_name, {}).get("score", 0) # Find best coin NOT currently active best_alternative = None for candidate in ranked: if candidate["name"] not in new_active: best_alternative = candidate break if best_alternative is None: continue gap = best_alternative["score"] - active_score if gap >= SWAP_SCORE_THRESHOLD: # Make the swap old_name = active_name new_name = best_alternative["name"] new_active.remove(old_name) new_active.append(new_name) # Initialize position tracking for new coin positions[new_name] = init_position(new_name) swap_record = { "time": datetime.now().isoformat(), "removed": old_name, "added": new_name, "old_score": active_score, "new_score": best_alternative["score"], "gap": gap, "old_reasons": scores[old_name].get("warnings", []), "new_reasons": best_alternative["reasons"], } swap_log.append(swap_record) swaps_made.append(swap_record) log.info(f"\n 🔄 SWAP: {old_name} (score {active_score}) → {new_name} (score {best_alternative['score']})") log.info(f" Removing {old_name}: {', '.join(scores[old_name].get('warnings', ['low score']))}") log.info(f" Adding {new_name}: {', '.join(best_alternative['reasons'][:3])}") notify( f"Coin Swap: {old_name} -> {new_name}", f"{old_name} score:{active_score} replaced by {new_name} score:{best_alternative['score']}\n" f"Reason: {', '.join(best_alternative['reasons'][:2])}" ) else: log.info(f" [{active_name}] Keeping — score {active_score} (best alt {best_alternative['name']} only {gap}pts better, need {SWAP_SCORE_THRESHOLD}+)") # Apply swaps ACTIVE_COINS = new_active if not swaps_made: log.info("\n No swaps needed — current coins are still optimal") else: log.info(f"\n {len(swaps_made)} swap(s) made") log.info(f" Active coins: {', '.join(ACTIVE_COINS)}") log.info("=" * 60) # Save swap log save_trades() last_optimize_date = datetime.now().date() return swaps_made # ── Trading Logic ───────────────────────────────────────────────────────────── def check_coin(name: str): global total_profit if name not in CANDIDATE_POOL: return config = CANDIDATE_POOL[name] pair = config["pair"] candles = get_ohlc(pair, interval=CANDLE_INTERVAL) if len(candles) < RSI_PERIOD + MA_PERIOD + 5: log.warning(f"[{name}] Not enough data ({len(candles)} candles)") return closes = [float(c[4]) for c in candles] current = closes[-1] rsi = calc_rsi(closes, RSI_PERIOD) ma = calc_ma(closes, MA_PERIOD) atr = calc_atr(candles, 14) atr_pct = (atr / current) * 100 pos = positions[name] log.info(f"[{name}] ${current:.4f} | RSI:{rsi:.1f} | MA:{ma:.4f} | ATR:{atr_pct:.2f}%") # ── Exit logic ──────────────────────────────────────────────────────────── if pos["active"]: pnl_pct = (current - pos["buy_price"]) / pos["buy_price"] pnl_usd = pnl_pct * (pos["volume"] * pos["buy_price"]) log.info(f"[{name}] Position P&L: {pnl_pct*100:+.2f}% (${pnl_usd:+.2f})") should_sell = False reason = "" if pnl_pct >= TAKE_PROFIT_PCT: should_sell, reason = True, f"TAKE PROFIT +{pnl_pct*100:.2f}%" elif pnl_pct <= -STOP_LOSS_PCT: should_sell, reason = True, f"STOP LOSS {pnl_pct*100:.2f}%" elif rsi >= RSI_OVERBOUGHT: should_sell, reason = True, f"RSI OVERBOUGHT ({rsi:.1f})" if should_sell: fees = (pos["buy_price"] * pos["volume"] * 0.0025) + (current * pos["volume"] * 0.0025) net = (current - pos["buy_price"]) * pos["volume"] - fees total_profit += net place_order("sell", pair, pos["volume"], round(current, config["decimals"])) trade_log.append({ "time": datetime.now().isoformat(), "coin": name, "pair": pair, "buy_price": pos["buy_price"], "sell_price": current, "volume": pos["volume"], "profit_usd": round(net, 4), "reason": reason, }) save_trades() log.info(f"[{name}] SELL @ ${current:.4f} | {reason} | Net: ${net:+.4f} | Total: ${total_profit:.4f}") notify(f"{name} SOLD — {reason}", f"Net: ${net:+.2f} | Total P&L: ${total_profit:.2f}") pos["active"] = False pos["buy_price"] = 0.0 pos["volume"] = 0.0 pos["order_id"] = None return # ── Entry logic ─────────────────────────────────────────────────────────── trend_ok = current > ma * 0.90 rsi_signal = rsi <= RSI_OVERSOLD vol_ok = atr_pct >= 0.5 if rsi_signal and trend_ok and vol_ok: volume = round(CAPITAL_PER_COIN / current, 6) if volume < config["min_order"]: log.warning(f"[{name}] Order size {volume:.4f} below minimum {config['min_order']}") return oid = place_order("buy", pair, volume, round(current, config["decimals"])) pos.update({ "active": True, "buy_price": current, "volume": volume, "order_id": oid, "buy_time": datetime.now().isoformat(), }) tp = round(current * (1 + TAKE_PROFIT_PCT), config["decimals"]) sl = round(current * (1 - STOP_LOSS_PCT), config["decimals"]) log.info(f"[{name}] BUY @ ${current:.4f} | RSI:{rsi:.1f} | Vol:{volume:.4f} | TP:${tp} | SL:${sl}") notify(f"{name} BUY — RSI {rsi:.0f}", f"${volume:.4f} @ ${current:.4f}\nTP:${tp} | SL:${sl}") else: reasons = [] if not rsi_signal: reasons.append(f"RSI {rsi:.1f} (need <{RSI_OVERSOLD})") if not trend_ok: reasons.append(f"below MA") if not vol_ok: reasons.append(f"low ATR {atr_pct:.2f}%") log.info(f"[{name}] No signal — {', '.join(reasons)}") # ── Persistence ─────────────────────────────────────────────────────────────── TRADES_FILE = "rsi_trades.json" def save_trades(): try: with open(TRADES_FILE, "w") as f: json.dump({ "total_profit": round(total_profit, 4), "trades": trade_log, "positions": positions, "active_coins": ACTIVE_COINS, "swap_log": swap_log, "last_optimized": str(last_optimize_date), }, f, indent=2, default=str) except Exception as e: log.error(f"save_trades failed: {e}") def load_trades(): global total_profit, trade_log, ACTIVE_COINS, last_optimize_date try: if os.path.exists(TRADES_FILE): with open(TRADES_FILE) as f: data = json.load(f) total_profit = data.get("total_profit", 0.0) trade_log = data.get("trades", []) saved_pos = data.get("positions", {}) saved_active = data.get("active_coins", ACTIVE_COINS) last_date_str = data.get("last_optimized") swap_log.extend(data.get("swap_log", [])) ACTIVE_COINS = saved_active for name in ACTIVE_COINS: positions[name] = init_position(name) if name in saved_pos: positions[name].update(saved_pos[name]) if last_date_str and last_date_str != "None": last_optimize_date = datetime.strptime(last_date_str, "%Y-%m-%d").date() log.info(f"Loaded {len(trade_log)} trades | P&L: ${total_profit:.4f} | " f"Active: {', '.join(ACTIVE_COINS)} | Swaps: {len(swap_log)}") except Exception as e: log.warning(f"Could not load trades: {e}") # ── Main loop ───────────────────────────────────────────────────────────────── def main(): global last_optimize_date mode = "DRY RUN" if DRY_RUN else "LIVE" log.info("=" * 60) log.info(f"RSI Mean Reversion Bot v2 — Self-Optimizing | {mode}") log.info(f"Starting coins: {', '.join(ACTIVE_COINS)}") log.info(f"Capital: ${TOTAL_CAPITAL:.0f} | Per coin: ${CAPITAL_PER_COIN:.2f}") log.info(f"Candidate pool: {len(CANDIDATE_POOL)} coins") log.info(f"Daily optimization at: {OPTIMIZE_HOUR:02d}:00") log.info(f"Swap threshold: {SWAP_SCORE_THRESHOLD}+ point advantage") log.info("=" * 60) load_trades() # Run optimization immediately on first start log.info("Running initial coin scan...") run_daily_optimization() cycle = 0 while True: cycle += 1 now = datetime.now() # ── Daily optimization check ───────────────────────────────────────── if (now.hour == OPTIMIZE_HOUR and now.minute < 2 and last_optimize_date != now.date()): run_daily_optimization() log.info(f"\n--- Cycle {cycle} | {now.strftime('%Y-%m-%d %H:%M:%S')} | " f"Active: {', '.join(ACTIVE_COINS)} ---") # ── Check each active coin ─────────────────────────────────────────── for name in list(ACTIVE_COINS): try: check_coin(name) time.sleep(1) except Exception as e: log.error(f"[{name}] Error: {e}") # ── Periodic summary ───────────────────────────────────────────────── if cycle % 10 == 0: active_pos = [n for n in ACTIVE_COINS if positions.get(n, {}).get("active")] log.info(f"\nSUMMARY | P&L: ${total_profit:.4f} | " f"Trades: {len(trade_log)} | " f"Swaps: {len(swap_log)} | " f"Open: {active_pos or 'none'}") save_trades() time.sleep(CHECK_INTERVAL) if __name__ == "__main__": try: main() except KeyboardInterrupt: log.info("\nBot stopped.") save_trades()