""" BTC 15-Minute Late Entry Bot Enter in last 2 minutes, avoid bad signals """ import os import json import time from datetime import datetime os.environ['KALSHI_API_KEY_ID'] = '81fe00fc-27b2-48ca-85f8-9c37ec1dc82f' os.environ['KALSHI_PRIVATE_KEY_PATH'] = './kalshi_private_key' from pykalshi import KalshiClient # Config PAPER_BALANCE = 1500 TARGET = 0.20 # 20% profit STOP = 0.15 # 15% stop POSITION = 0.20 # 20% per trade class BTC15MLateEntry: def __init__(self): self.client = None self.balance = PAPER_BALANCE self.positions = [] self.trades = [] self.prices = {} # ticker -> [(time, price), ...] def setup(self): self.client = KalshiClient() print("โœ… BTC 15m Bot Connected") return self def get_markets(self): markets = self.client.get_markets(series_ticker='KXBTC15M') return [m for m in markets if m.volume > 0] def price(self, market): if market.yes_bid == 0 and market.yes_ask == 0: return 0 return (market.yes_bid + market.yes_ask) / 200 def analyze_2min_entry(self, ticker, current_price): """ Late Entry Strategy - Enter in last 2 minutes with good signals Rules: 1. Must be in last 2 minutes of the 15-min window 2. Avoid long wicks (sudden spikes) 3. Look for stable momentum (not choppy) 4. Enter with the trend """ now = datetime.now() if ticker not in self.prices: self.prices[ticker] = [] # Add current price self.prices[ticker].append((now, current_price)) # Keep last 3 minutes cutoff = now.timestamp() - 180 self.prices[ticker] = [(t, p) for t, p in self.prices[ticker] if t.timestamp() > cutoff] if len(self.prices[ticker]) < 3: return None prices = [p for _, p in self.prices[ticker]] # Get timing info from ticker # Format: KXBTC15M-26FEB191915-15 try: parts = ticker.split('-') time_str = parts[2] + parts[3] # e.g., 191915 minute = int(time_str[4:6]) second = int(time_str[6:8]) # Convert to total seconds from hour hour = int(time_str[2:4]) current_minute = minute # This is the expiration minute current_second = second # We want to enter in last 2 minutes (120 seconds) time_remaining = current_minute * 60 - (now.minute * 60 + now.second) except: time_remaining = 999 # Unknown timing # If not in last 2 minutes, skip if time_remaining > 120: return None # Analyze price action first_price = prices[0] last_price = prices[-1] # Calculate metrics change = last_price - first_price volatility = max(prices) - min(prices) # Avoid signals with: # 1. High volatility (long wicks) - volatility > 15% is suspicious # 2. Choppy - if price keeps flipping direction # Count direction changes direction_changes = 0 for i in range(1, len(prices)): if (prices[i] - prices[i-1]) > 0.02: direction = 'up' elif (prices[i] - prices[i-1]) < -0.02: direction = 'down' else: direction = 'flat' if i > 1 and direction != prev_dir and direction != 'flat': direction_changes += 1 prev_dir = direction # Good signal criteria: # 1. Price trending in one direction (not choppy) # 2. Not too volatile # 3. Move is at least 3% if direction_changes <= 1 and volatility < 0.15: if change > 0.03: return 'YES' # Price going up elif change < -0.03: return 'NO' # Price going down return None def enter(self, market, side, price): contracts = int(self.balance * POSITION) if contracts < 1: return pos = { 'ticker': market.ticker, 'side': side, 'entry': price, 'contracts': contracts, 'time': datetime.now().isoformat(), 'status': 'open' } self.positions.append(pos) self.balance -= contracts print(f"\n๐Ÿ“ 2-MIN ENTRY {side} @ {price*100:.0f}ยข") print(f" {market.ticker[-20:]}") print(f" ${contracts} | Balance: ${self.balance:.0f}") def check_exits(self, markets): market_dict = {m.ticker: m for m in markets} for p in self.positions: if p['status'] != 'open': continue if p['ticker'] not in market_dict: continue m = market_dict[p['ticker']] curr = self.price(m) if p['side'] == 'YES': pnl_pct = (curr - p['entry']) / p['entry'] if p['entry'] > 0 else 0 else: pnl_pct = (p['entry'] - curr) / p['entry'] if p['entry'] > 0 else 0 if pnl_pct >= TARGET: self.close(p, curr, "TARGET") elif pnl_pct <= -STOP: self.close(p, curr, "STOP") def close(self, pos, price, reason): pnl = (price - pos['entry']) * pos['contracts'] if pos['side'] == 'YES' else (pos['entry'] - price) * pos['contracts'] self.balance += pos['contracts'] + pnl pos['status'] = 'closed' pos['pnl'] = pnl pos['exit'] = price pos['reason'] = reason self.trades.append(pos) if pnl > 0: print(f"\nโœ… WIN: +${pnl:.0f}") else: print(f"\nโŒ LOSS: ${pnl:.0f}") print(f" Balance: ${self.balance:.0f}") def run(self, cycles=30, delay=2): print(f"\n๐Ÿš€ BTC 15m Late Entry Bot") print(f" Balance: ${PAPER_BALANCE}") print(f" Target: +{TARGET*100:.0f}% | Stop: -{STOP*100:.0f}%") print(f" Strategy: Enter in last 2 minutes, avoid wicks\n") for i in range(cycles): markets = self.get_markets() # Check exits self.check_exits(markets) # Check entries - Late Entry for m in markets: if any(p['status'] == 'open' and p['ticker'] == m.ticker for p in self.positions): continue curr = self.price(m) if curr == 0: continue signal = self.analyze_2min_entry(m.ticker, curr) if signal: self.enter(m, signal, curr) # Stats closed = [t for t in self.trades if t['status'] == 'closed'] wins = len([t for t in closed if t.get('pnl', 0) > 0]) open_pos = len([p for p in self.positions if p['status'] == 'open']) print(f"[{i+1}] Active: {len(markets)} | Open: {open_pos} | Balance: ${self.balance:.0f}") time.sleep(delay) # Final self.print_stats() with open('btc15m_late_results.json', 'w') as f: json.dump({ 'balance': self.balance, 'trades': self.trades, 'positions': self.positions }, f, indent=2, default=str) def print_stats(self): closed = [t for t in self.trades if t['status'] == 'closed'] if closed: wins = [t for t in closed if t.get('pnl', 0) > 0] print(f"\n{'='*50}") print(f"FINAL: ${self.balance:.2f}") print(f"Trades: {len(closed)} | Wins: {len(wins)} | Win Rate: {len(wins)/len(closed)*100:.0f}%") if __name__ == '__main__': BTC15MLateEntry().setup().run(cycles=30, delay=2)