""" Late Entry BTC 15-Minute Bot Enter in last 2 minutes based on momentum """ 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, MarketStatus # Config PAPER_BALANCE = 1500 MIN_PROFIT = 0.15 # 15% profit target MAX_LOSS = 0.10 # 10% stop POSITION_PCT = 0.20 # 20% per trade class BTC15Bot: def __init__(self): self.client = None self.balance = PAPER_BALANCE self.positions = [] self.trades = [] self.losses = 0 self.prices = {} # ticker -> list of (time, price) def setup(self): self.client = KalshiClient() print("โœ… Connected to Kalshi") return self def get_markets(self): """Get active BTC 15-min markets""" try: markets = self.client.get_markets( series_ticker='KXBTC15M', limit=20 ) # Filter for markets with volume return [m for m in markets if m.volume > 0] except Exception as e: print(f"Error: {e}") return [] def get_current_price(self, market): """Get YES price midpoint""" if market.yes_bid == 0 and market.yes_ask == 0: return 0 return (market.yes_bid + market.yes_ask) / 200 def get_market_time(self, ticker): """Extract time from ticker to know when it expires""" # Ticker format: KXBTC15M-26FEB191915-15 # Last part is the minute (15 = 15 minutes) try: parts = ticker.split('-') time_str = parts[-2] + parts[-1] # e.g., 191915 # Parse as HHMMSS hour = int(time_str[2:4]) minute = int(time_str[4:6]) return hour, minute except: return None, None def analyze_signal(self, ticker, current_price): """Late Entry: Enter in last 2 minutes based on momentum""" if current_price == 0: return None now = datetime.now() if ticker not in self.prices: self.prices[ticker] = [] # Add current price self.prices[ticker].append((now, current_price)) # Keep only 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]) < 2: return None # Get price change first_price = self.prices[ticker][0][1] price_change = current_price - first_price # Late Entry: If price trending up in last few minutes, bet YES # If trending down, bet NO if price_change > 0.03: # 3%+ move up return 'YES' elif price_change < -0.03: # 3%+ move down return 'NO' return None def enter(self, market, side, price): """Enter a trade""" contracts = int(self.balance * POSITION_PCT) if contracts < 1: return pos = { 'ticker': market.ticker, 'side': side, 'entry': price, 'contracts': contracts, 'entry_time': datetime.now().isoformat(), 'status': 'open' } self.positions.append(pos) self.balance -= contracts print(f"\n๐Ÿ“ ENTER {side} @ {price*100:.0f}ยข") print(f" {market.ticker[-20:]}") print(f" ${contracts} | Balance: ${self.balance:.0f}") def check_exits(self, markets): """Check for exit signals""" market_dict = {m.ticker: m for m in markets} for pos in self.positions: if pos['status'] != 'open': continue if pos['ticker'] not in market_dict: continue market = market_dict[pos['ticker']] current = self.get_current_price(market) if pos['side'] == 'YES': pnl_pct = (current - pos['entry']) / pos['entry'] if pos['entry'] > 0 else 0 else: pnl_pct = (pos['entry'] - current) / pos['entry'] if pos['entry'] > 0 else 0 # Exit conditions if pnl_pct >= MIN_PROFIT: self.close(pos, current, f"PROFIT +{pnl_pct*100:.0f}%") elif pnl_pct <= -MAX_LOSS: self.close(pos, current, f"STOP {pnl_pct*100:.0f}%") def close(self, pos, price, reason): """Close position""" if pos['side'] == 'YES': pnl = (price - pos['entry']) * pos['contracts'] else: pnl = (pos['entry'] - price) * pos['contracts'] self.balance += pos['contracts'] + pnl pos['status'] = 'closed' pos['exit'] = price pos['pnl'] = pnl pos['reason'] = reason pos['exit_time'] = datetime.now().isoformat() self.trades.append(pos) if pnl > 0: self.losses = 0 print(f"\nโœ… WIN: {reason} | +${pnl:.0f}") else: self.losses += 1 print(f"\nโŒ LOSS: {reason} | ${pnl:.0f}") print(f" Balance: ${self.balance:.0f}") def run(self, cycles=30, delay=2): """Main loop""" print(f"\n๐Ÿš€ BTC 15-min Bot Running...") print(f" Balance: ${PAPER_BALANCE}") print(f" Target: +{MIN_PROFIT*100:.0f}% | Stop: -{MAX_LOSS*100:.0f}%") print() for i in range(cycles): markets = self.get_markets() # Check exits self.check_exits(markets) if self.losses >= 3: print("๐Ÿ›‘ Circuit breaker") break # Check entries for m in markets: # Skip if already have position if any(p['status'] == 'open' and p['ticker'] == m.ticker for p in self.positions): continue price = self.get_current_price(m) if price == 0: continue signal = self.analyze_signal(m.ticker, price) if signal: self.enter(m, signal, price) # Stats closed = [t for t in self.trades if t['status'] == 'closed'] wins = len([t for t in closed if t['pnl'] > 0]) open_pos = len([p for p in self.positions if p['status'] == 'open']) print(f"[{i+1}] Balance: ${self.balance:.0f} | Trades: {len(closed)} | Wins: {wins} | Open: {open_pos}") time.sleep(delay) # Final stats print("\n" + "="*50) print(f"FINAL BALANCE: ${self.balance:.2f}") closed = [t for t in self.trades if t['status'] == 'closed'] if closed: wins = [t for t in closed if t['pnl'] > 0] print(f"Trades: {len(closed)} | Wins: {len(wins)} | Win Rate: {len(wins)/len(closed)*100:.0f}%") # Save with open('btc15m_bot_state.json', 'w') as f: json.dump({ 'balance': self.balance, 'trades': self.trades, 'positions': self.positions }, f, indent=2, default=str) print("๐Ÿ’พ Saved") if __name__ == '__main__': BTC15Bot().setup().run(cycles=15, delay=2)