""" PAPER TRADING BOT - BTC 15-Min Kalshi Markets $100 starting balance. Smart tiered bet sizing. 2:1 reward/risk. Trailing high-water mark stop-loss — protects profits, not just starting balance. Live dashboard: open dashboard.html via OPEN DASHBOARD shortcut """ import os, base64, datetime, time, json, threading, requests from dotenv import load_dotenv from cryptography.hazmat.primitives import serialization, hashes from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.backends import default_backend try: import websocket WS_AVAILABLE = True except ImportError: WS_AVAILABLE = False load_dotenv() KEY_ID = os.getenv("KALSHI_API_KEY_ID") KEY_PATH = os.getenv("KALSHI_PRIVATE_KEY_PATH") BASE_URL = "https://api.elections.kalshi.com" STARTING_BALANCE = 100.00 PROFIT_TARGET = 0.40 # exit when contract gains 40% STOP_LOSS = 0.25 # exit when contract loses 25% (wider = less noise stops) MAX_ENTRY_PRICE = 0.75 MIN_ENTRY_PRICE = 0.15 MOMENTUM_SECONDS = 120 MOMENTUM_THRESHOLD = 0.0010 # 0.1% BTC move required — filters noise BALANCED_LOW = 0.30 BALANCED_HIGH = 0.70 MAX_POSITIONS = 2 # max 2 open at once (down from 3) SCAN_INTERVAL = 10 MIN_SECS_LEFT = 300 # only enter with 5+ min left WARMUP_TICKS = 4 TRAILING_STOP_PCT = 0.20 BET_PCT_OF_BALANCE = 0.02 # 2% of balance per trade (down from 5%) BET_MAX_ABS = 2.00 # max $2 per trade (was $5) BET_MIN_ABS = 0.75 # min $0.75 ENTRY_COOLDOWN = 90 # 90 sec between entries prices = {"binance": None, "coinbase": None, "kraken": None} price_times = {"binance": None, "coinbase": None, "kraken": None} price_history = [] price_lock = threading.Lock() paper_positions = [] paper_log = [] session_start = time.time() balance = STARTING_BALANCE high_water_mark = STARTING_BALANCE session_halted = False traded_tickers = set() # prevent re-entering same market ticker position_lock = threading.Lock() # protects paper_positions + balance across threads FAST_MONITOR_INTERVAL = 2 # check open positions every 2 seconds def load_private_key(path): with open(path, "rb") as f: return serialization.load_pem_private_key(f.read(), password=None, backend=default_backend()) def sign_request(pk, ts, method, path): msg = (ts + method + path).encode("utf-8") sig = pk.sign(msg, padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.DIGEST_LENGTH), hashes.SHA256()) return base64.b64encode(sig).decode("utf-8") def get_headers(method, path): ts = str(int(datetime.datetime.now().timestamp() * 1000)) pk = load_private_key(KEY_PATH) return {"KALSHI-ACCESS-KEY": KEY_ID, "KALSHI-ACCESS-TIMESTAMP": ts, "KALSHI-ACCESS-SIGNATURE": sign_request(pk, ts, method, path), "Content-Type": "application/json"} def update_high_water_mark(): global high_water_mark if balance > high_water_mark: print(f" ** NEW HIGH: ${balance:.2f} Stop now at ${balance*(1-TRAILING_STOP_PCT):.2f} **") high_water_mark = balance def check_session_stop_loss(): global session_halted if session_halted: return True stop_level = high_water_mark * (1 - TRAILING_STOP_PCT) if balance <= stop_level: session_halted = True print(f"\n !! TRAILING STOP HIT — Peak:${high_water_mark:.2f} Now:${balance:.2f} Stop:${stop_level:.2f} !!") print(f" !! Trading halted — monitoring open positions only !!\n") return True return False def binance_ws(): if not WS_AVAILABLE: return def on_msg(ws, msg): try: data = json.loads(msg) price = float(data.get("p") or data.get("c") or 0) if price > 0: now = time.time() with price_lock: prices["binance"] = price price_times["binance"] = now price_history.append((now, price)) except: pass def on_error(ws, err): pass def on_close(ws, *args): time.sleep(5) try: binance_ws() except: pass try: ws = websocket.WebSocketApp("wss://stream.binance.com:9443/ws/btcusdt@trade", on_message=on_msg, on_error=on_error, on_close=on_close) ws.run_forever(ping_interval=30, ping_timeout=10) except: pass def poll_binance_rest(): while True: try: r = requests.get("https://api.binance.com/api/v3/ticker/price?symbol=BTCUSDT", timeout=5) if r.status_code == 200: price = float(r.json()["price"]) now = time.time() with price_lock: prices["binance"] = price; price_times["binance"] = now price_history.append((now, price)) except: pass time.sleep(10) def poll_coinbase(): while True: try: r = requests.get("https://api.coinbase.com/v2/prices/BTC-USD/spot", timeout=5) if r.status_code == 200: price = float(r.json()["data"]["amount"]) now = time.time() with price_lock: prices["coinbase"] = price; price_times["coinbase"] = now price_history.append((now, price)) except: pass time.sleep(5) def poll_kraken(): while True: try: r = requests.get("https://api.kraken.com/0/public/Ticker?pair=XBTUSD", timeout=5) if r.status_code == 200: result = r.json().get("result", {}) price = float(result.get("XXBTZUSD", {}).get("c", [0])[0]) if price > 0: now = time.time() with price_lock: prices["kraken"] = price; price_times["kraken"] = now price_history.append((now, price)) except: pass time.sleep(8) def get_best_price(): with price_lock: snap_p = dict(prices); snap_t = dict(price_times) now = time.time() valid = [(src, snap_p[src]) for src in snap_p if snap_p[src] and snap_t[src] and (now - snap_t[src]) < 30] if not valid: return None, None, None price = snap_p["binance"] or snap_p["coinbase"] or valid[0][1] srcs = "+".join(s for s, _ in valid) edge = None b, c, k = snap_p["binance"], snap_p["coinbase"], snap_p["kraken"] if b and c and k: avg = (c + k) / 2; div = (b - avg) / avg if abs(div) > 0.0005: edge = f"EDGE {div*100:+.3f}% Binance vs others" return price, srcs, edge def get_momentum(): now = time.time(); cutoff = now - MOMENTUM_SECONDS with price_lock: recent = [(t, p) for t, p in price_history if t >= cutoff] price_history[:] = [(t, p) for t, p in price_history if t >= now - 600] if len(recent) < WARMUP_TICKS: return None, None, None oldest = recent[0][1]; newest = recent[-1][1] mom = (newest - oldest) / oldest pct = f"{mom*100:+.3f}%" if mom >= MOMENTUM_THRESHOLD: return mom, "UP", pct elif mom <= -MOMENTUM_THRESHOLD: return mom, "DOWN", pct return mom, "FLAT", pct def get_signal(market, momentum, direction): """MOMENTUM only — no BALANCED. Require real directional move.""" if not market or momentum is None: return False, None, None if direction in ("UP", "DOWN"): return True, direction, "MOMENTUM" return False, None, None def get_bet_size(momentum, edge, signal_type): base_bet = round(balance * BET_PCT_OF_BALANCE, 2) if signal_type == "BALANCED": bet, tier = round(base_bet * 0.50, 2), "BALANCED" if edge: bet, tier = round(min(bet * 1.5, base_bet), 2), "BALANCED+EDGE" else: strength = abs(momentum or 0) / MOMENTUM_THRESHOLD if strength < 2.0: mult, tier = 0.50, "WEAK" elif strength < 4.0: mult, tier = 0.75, "BASE" elif strength < 6.0: mult, tier = 1.00, "MEDIUM" else: mult, tier = 1.50, "STRONG" bet = round(base_bet * mult, 2) if edge and strength >= 2.0: bet = round(min(bet * 1.25, BET_MAX_ABS), 2) tier = tier + "+EDGE" bet = max(BET_MIN_ABS, min(bet, BET_MAX_ABS, balance * 0.10)) return round(bet, 2), tier def get_15m_market(): path = "/trade-api/v2/markets" params = "?status=open&limit=10&series_ticker=KXBTC15M" headers = get_headers("GET", path) r = requests.get(BASE_URL + path + params, headers=headers) if r.status_code != 200: return None now = datetime.datetime.now(datetime.timezone.utc) for m in r.json().get("markets", []): try: open_t = datetime.datetime.fromisoformat(m["open_time"].replace("Z", "+00:00")) close_t = datetime.datetime.fromisoformat(m["close_time"].replace("Z", "+00:00")) if open_t <= now <= close_t: secs = (close_t - now).total_seconds() return {"ticker": m["ticker"], "yes_ask": float(m.get("yes_ask_dollars") or 0), "yes_bid": float(m.get("yes_bid_dollars") or 0), "no_ask": float(m.get("no_ask_dollars") or 0), "no_bid": float(m.get("no_bid_dollars") or 0), "floor": float(m.get("floor_strike") or 0), "secs_left": secs, "volume": float(m.get("volume_24h_fp") or 0)} except: continue return None def get_live_price(ticker): path = f"/trade-api/v2/markets/{ticker}" headers = get_headers("GET", path) r = requests.get(BASE_URL + path, headers=headers) if r.status_code == 200: m = r.json().get("market", {}) return (float(m.get("yes_bid_dollars") or 0), float(m.get("no_bid_dollars") or 0)) return None, None # ── Auto-analysis thresholds (only tune on VERY strong patterns) ────────────── AUTO_MIN_BUCKET_TRADES = 20 # need 20+ trades in a bucket before acting AUTO_MIN_WIN_RATE = 0.60 # 60%+ win rate required to mark as "good" AUTO_MAX_WIN_RATE = 0.40 # 40% or below to mark as "avoid" AUTO_RETUNE_EVERY = 30 # re-analyze every 30 completed trades _last_autotune_count = 0 # track when we last auto-tuned def get_btc_volatility(): """Std deviation of last 60 BTC ticks — measures choppiness.""" with price_lock: recent = [p for _, p in price_history[-60:]] if len(recent) < 10: return 0.0 mean = sum(recent) / len(recent) variance = sum((p - mean) ** 2 for p in recent) / len(recent) return round((variance ** 0.5) / mean * 100, 4) # as % of price def log_analytics(pos, event, exit_price=None, pnl=None, exit_reason=None): """Write rich per-trade data to trade_analytics.json for the learning engine.""" now = datetime.datetime.now() record = { "event" : event, "ticker" : pos["ticker"], "timestamp" : time.time(), "time_str" : now.strftime("%H:%M:%S"), "hour" : now.hour, "day_of_week" : now.weekday(), # 0=Mon 6=Sun "direction" : pos["direction"], "signal_type" : pos.get("signal_type", "?"), "tier" : pos.get("tier", "?"), "entry_price" : pos["entry_price"], "momentum_pct" : round(abs(pos.get("momentum_raw", 0)) * 100, 4), "momentum_strength": round(abs(pos.get("momentum_raw", 0)) / MOMENTUM_THRESHOLD, 2), "btc_volatility" : get_btc_volatility(), "market_yes_ask" : pos.get("market_yes_ask", 0), "market_spread" : round(pos.get("market_yes_ask", 0) - pos.get("market_yes_bid", 0), 3), "secs_left" : pos.get("secs_left", 0), "btc_price" : pos.get("entry_btc", 0), } if event == "exit" and exit_price is not None: hold_secs = int(time.time() - pos.get("entry_time", time.time())) record.update({ "exit_price" : exit_price, "pnl" : round(pnl, 4), "pnl_pct" : round((exit_price - pos["entry_price"]) / pos["entry_price"] * 100, 2), "outcome" : "WIN" if (pnl or 0) > 0 else "LOSS", "exit_reason": exit_reason, "hold_secs" : hold_secs, }) # Append to analytics file analytics_path = "C:\\kalshibot\\trade_analytics.json" try: try: with open(analytics_path) as f: data = json.load(f) except: data = [] data.append(record) with open(analytics_path, "w") as f: json.dump(data, f, indent=2, default=str) except Exception as e: pass # never crash the bot over analytics def log_market_snapshot(btc_price, market, momentum, direction): """ Runs every scan cycle. Tracks BTC price trend + Kalshi odds over time. Stored in market_snapshots.json — separate from per-trade analytics. """ if not btc_price: return now = datetime.datetime.now() # Calculate short-term BTC change vs last snapshot snap_path = "C:\\kalshibot\\market_snapshots.json" try: try: with open(snap_path) as f: snaps = json.load(f) except: snaps = [] # BTC change vs 1 min ago and 5 min ago now_ts = time.time() snaps_1m = [s for s in snaps if s.get("timestamp") and (now_ts - s["timestamp"]) <= 65] snaps_5m = [s for s in snaps if s.get("timestamp") and (now_ts - s["timestamp"]) <= 305] btc_1m_ago = snaps_1m[0]["btc_price"] if snaps_1m else btc_price btc_5m_ago = snaps_5m[0]["btc_price"] if snaps_5m else btc_price chg_1m = round((btc_price - btc_1m_ago) / btc_1m_ago * 100, 4) chg_5m = round((btc_price - btc_5m_ago) / btc_5m_ago * 100, 4) record = { "timestamp" : now_ts, "time_str" : now.strftime("%H:%M:%S"), "hour" : now.hour, "day_of_week": now.weekday(), "btc_price" : btc_price, "btc_chg_1m" : chg_1m, "btc_chg_5m" : chg_5m, "momentum" : round(momentum * 100, 4) if momentum else 0, "direction" : direction or "FLAT", "volatility" : get_btc_volatility(), "yes_ask" : market["yes_ask"] if market else None, "no_ask" : market["no_ask"] if market else None, "spread" : round(market["yes_ask"] - market["yes_bid"], 3) if market else None, "secs_left" : market["secs_left"] if market else None, } snaps.append(record) # Keep last 2000 snapshots (~5.5 hours at 10s intervals) if len(snaps) > 2000: snaps = snaps[-2000:] with open(snap_path, "w") as f: json.dump(snaps, f, default=str) except: pass def auto_analyze(): """ Runs every AUTO_RETUNE_EVERY completed trades. Only updates auto_params.json when patterns are very strong: - 60%+ win rate AND 20+ trades in bucket = mark as GOOD hour - 40% or below AND 20+ trades = mark as AVOID hour Does NOT constantly tweak — only acts on overwhelming evidence. """ global _last_autotune_count, avoid_hours, good_hours analytics_path = "C:\\kalshibot\\trade_analytics.json" auto_params_path = "C:\\kalshibot\\auto_params.json" try: with open(analytics_path) as f: data = json.load(f) except: return exits = [r for r in data if r.get("event") == "exit"] if len(exits) < AUTO_RETUNE_EVERY: return # ── Hour analysis ── from collections import defaultdict hour_map = defaultdict(lambda: {"wins": 0, "total": 0}) for t in exits: h = t.get("hour", 0) hour_map[h]["total"] += 1 if t.get("outcome") == "WIN": hour_map[h]["wins"] += 1 new_good = [h for h, d in hour_map.items() if d["total"] >= AUTO_MIN_BUCKET_TRADES and d["wins"] / d["total"] >= AUTO_MIN_WIN_RATE] new_avoid = [h for h, d in hour_map.items() if d["total"] >= AUTO_MIN_BUCKET_TRADES and d["wins"] / d["total"] <= AUTO_MAX_WIN_RATE] # ── Momentum strength analysis ── mom_map = defaultdict(lambda: {"wins": 0, "total": 0}) for t in exits: bucket = "strong" if t.get("momentum_strength", 0) >= 2.0 else "weak" mom_map[bucket]["total"] += 1 if t.get("outcome") == "WIN": mom_map[bucket]["wins"] += 1 # If weak signals are losing badly (40%- with 20+ trades), note it weak_note = "" if mom_map["weak"]["total"] >= AUTO_MIN_BUCKET_TRADES: weak_wr = mom_map["weak"]["wins"] / mom_map["weak"]["total"] if weak_wr <= AUTO_MAX_WIN_RATE: weak_note = f"Weak signals losing ({weak_wr*100:.0f}% WR) — consider raising threshold" # ── Only write if something actually changed ── existing_good = sorted(good_hours) existing_avoid = sorted(avoid_hours) new_good_s = sorted(new_good) new_avoid_s = sorted(new_avoid) changed = (new_good_s != existing_good or new_avoid_s != existing_avoid) notes = [] if new_good_s: notes.append(f"Best hours (60%+ WR, 20+ trades): {[f'{h:02d}:xx' for h in new_good_s]}") if new_avoid_s: notes.append(f"Avoided hours (40%- WR, 20+ trades): {[f'{h:02d}:xx' for h in new_avoid_s]}") if weak_note: notes.append(weak_note) params = { "generated_at" : str(datetime.datetime.now()), "trades_analyzed" : len(exits), "good_hours" : new_good_s, "avoid_hours" : new_avoid_s, "notes" : notes, } with open(auto_params_path, "w") as f: json.dump(params, f, indent=2) # Apply immediately in this session good_hours = new_good_s avoid_hours = new_avoid_s _last_autotune_count = len(exits) if changed and (new_good_s or new_avoid_s): print(f"\n [LEARN] Auto-params updated ({len(exits)} trades analyzed)") for note in notes: print(f" {note}") else: print(f" [LEARN] {len(exits)} trades analyzed — no strong patterns yet (need 60%+ or 40%- with 20+ trades)") def save_log(): with open("C:\\kalshibot\\paper_log.json", "w") as f: json.dump({"balance": balance, "high_water_mark": high_water_mark, "trades": paper_log}, f, indent=2, default=str) def save_state(btc, sources, edge, momentum, direction, market, bet_tier="BASE"): exits = [t for t in paper_log if t.get("event") == "exit"] # ── KEY CHANGE: include live current_price and computed P&L on each position ── pos_data = [] for p in paper_positions: current = p.get("current_price", p["entry_price"]) pnl_abs = round((current - p["entry_price"]) * p["count"], 4) pnl_pct = round((current - p["entry_price"]) / p["entry_price"] * 100, 2) held_secs = int(time.time() - p.get("entry_time", time.time())) pos_data.append({ "ticker" : p["ticker"], "side" : p["side"], "direction" : p["direction"], "signal_type" : p.get("signal_type", "?"), "tier" : p.get("tier", "?"), "entry_price" : p["entry_price"], "current_price": current, "target" : p["target"], "stop" : p["stop"], "count" : p["count"], "cost" : p["cost"], "pnl_abs" : pnl_abs, "pnl_pct" : pnl_pct, "floor" : p["floor"], "held_secs" : held_secs, "entry_time_str": datetime.datetime.fromtimestamp( p.get("entry_time", time.time())).strftime("%H:%M:%S"), }) stop_level = round(high_water_mark * (1 - TRAILING_STOP_PCT), 2) state = { "mode": "paper", "last_update": time.time(), "btc_price": btc, "btc_source": sources, "edge": edge or "", "balance": round(balance, 2), "starting_balance": STARTING_BALANCE, "high_water_mark": round(high_water_mark, 2), "stop_level": stop_level, "momentum": momentum or 0, "direction": direction or "FLAT", "bet_tier": bet_tier, "session_halted": session_halted, "market": market, "open_positions": pos_data, "completed_trades": exits } try: with open("C:\\kalshibot\\bot_state.json", "w") as f: json.dump(state, f, default=str) except: pass def paper_enter(market, trade_direction, signal_type, btc_price, momentum, edge): global balance ticker = market["ticker"] # NEVER enter same market twice if ticker in traded_tickers: print(f" SKIP — already traded {ticker[-15:]} this window") return False, "DUP" side = "yes" if trade_direction == "UP" else "no" entry_price = market["yes_ask"] if side == "yes" else market["no_ask"] if not (MIN_ENTRY_PRICE <= entry_price <= MAX_ENTRY_PRICE): return False, "SKIP" bet, tier = get_bet_size(momentum, edge, signal_type) bet = min(bet, balance * 0.10) count = max(1, int(bet / entry_price)) cost = round(entry_price * count, 2) if cost > balance: return False, tier target = round(entry_price * (1 + PROFIT_TARGET), 4) stop = round(entry_price * (1 - STOP_LOSS), 4) max_gain = round((target - entry_price) * count, 2) max_loss = round((entry_price - stop) * count, 2) rr = round(max_gain / max_loss, 1) if max_loss > 0 else 0 mins = int(market["secs_left"] // 60) secs_r = int(market["secs_left"] % 60) pos = {"ticker": market["ticker"], "side": side, "entry_price": entry_price, "current_price": entry_price, "count": count, "cost": cost, "target": target, "stop": stop, "max_gain": max_gain, "max_loss": max_loss, "entry_time": time.time(), "entry_btc": btc_price, "direction": trade_direction, "floor": market["floor"], "tier": tier, "signal_type": signal_type, "momentum_raw": momentum or 0, "market_yes_ask": market["yes_ask"], "market_yes_bid": market["yes_bid"], "secs_left": market["secs_left"]} balance -= cost with position_lock: paper_positions.append(pos) traded_tickers.add(ticker) # lock this market for rest of window paper_log.append({**pos, "event": "entry", "time_str": datetime.datetime.now().strftime("%H:%M:%S")}) log_analytics(pos, "entry") save_log() stop_level = round(high_water_mark * (1 - TRAILING_STOP_PCT), 2) print(f"\n {'='*62}") print(f" PAPER TRADE [{signal_type}] {'YES BTC UP' if side=='yes' else 'NO BTC DOWN'} [{tier}]") print(f" BTC: ${btc_price:,.2f} Floor: ${market['floor']:,.2f}") print(f" Entry: ${entry_price:.3f} x{count} = ${cost:.2f} | {mins}m {secs_r}s left") print(f" Target: +${max_gain:.2f} (${target:.3f}) Stop: -${max_loss:.2f} (${stop:.3f}) [{rr}:1]") print(f" Balance: ${balance:.2f} Peak: ${high_water_mark:.2f} Trailing stop: ${stop_level:.2f}") print(f" {'='*62}") return True, tier def check_exits_fast(): """ Called by the 2-second fast monitor thread. Checks every open position for target/stop/time exit. Prints only when something actually exits — silent otherwise. """ global balance with position_lock: if not paper_positions: return to_close = [] for pos in paper_positions: yb, nb = get_live_price(pos["ticker"]) if yb is None: continue current = yb if pos["side"] == "yes" else nb if current <= 0: continue pos["current_price"] = current entry = pos["entry_price"] pnl_pct = (current - entry) / entry pnl_abs = round((current - entry) * pos["count"], 4) reason = None if current >= pos["target"]: reason = f"TARGET +{pnl_pct*100:.1f}%" elif current <= pos["stop"]: reason = f"STOP {pnl_pct*100:.1f}%" elif time.time() - pos["entry_time"] > 780: reason = f"TIME {pnl_pct*100:+.1f}%" if reason: balance = round(balance + current * pos["count"], 2) icon = "WIN " if pnl_abs >= 0 else "LOSS" print(f"\n ⚡[{icon}] {reason} ${entry:.3f}→${current:.3f} " f"P&L:${pnl_abs:+.2f} Bal:${balance:.2f}") paper_log.append({**pos, "event": "exit", "exit_price": current, "pnl": pnl_abs, "reason": reason, "balance_after": balance, "time_str": datetime.datetime.now().strftime("%H:%M:%S")}) log_analytics(pos, "exit", exit_price=current, pnl=pnl_abs, exit_reason=reason) to_close.append(pos) save_log() for pos in to_close: paper_positions.remove(pos) def fast_monitor_loop(): """Daemon thread — checks positions every 2 seconds independently of the main loop.""" while True: try: time.sleep(FAST_MONITOR_INTERVAL) check_exits_fast() except Exception as e: pass # never kill the thread on an error def print_position_status(): """Called by main loop every 10 seconds — just prints HOLD status, no exit logic.""" with position_lock: for pos in paper_positions: current = pos.get("current_price", pos["entry_price"]) entry = pos["entry_price"] pnl_pct = (current - entry) / entry print(f" HOLD {'YES' if pos['side']=='yes' else 'NO '} " f"${entry:.3f}→${current:.3f} ({pnl_pct*100:+.1f}%) " f"target:${pos['target']:.3f} stop:${pos['stop']:.3f}") def paper_monitor(): """Legacy alias kept so nothing breaks — calls fast exit checker.""" check_exits_fast() def print_report(): print("\n\n" + "="*65) print(" PAPER TRADING SESSION REPORT") print("="*65) exits = [t for t in paper_log if t.get("event") == "exit"] if not exits: print(f" No completed trades. Balance: ${balance:.2f} Peak: ${high_water_mark:.2f}") return wins = [t for t in exits if t.get("pnl", 0) > 0] losses = [t for t in exits if t.get("pnl", 0) <= 0] total_pnl = round(sum(t.get("pnl", 0) for t in exits), 2) total_bet = round(sum(t.get("cost", 0) for t in exits), 2) hrs = int((time.time() - session_start) / 3600) mins = int(((time.time() - session_start) % 3600) / 60) print(f" Session : {hrs}h {mins}m") print(f" Start : ${STARTING_BALANCE:.2f}") print(f" Peak balance : ${high_water_mark:.2f} (+${high_water_mark-STARTING_BALANCE:.2f})") print(f" Final balance: ${balance:.2f}") print(f" Total P&L : ${total_pnl:+.2f} ({(balance-STARTING_BALANCE)/STARTING_BALANCE*100:+.1f}%)") print(f" Total wagered: ${total_bet:.2f}") print(f" {'─'*45}") print(f" Trades : {len(exits)} ({len(wins)}W / {len(losses)}L) Win rate: {len(wins)/len(exits)*100:.0f}%") if exits: print(f" Avg/trade : ${total_pnl/len(exits):+.2f}") types = {} for t in exits: st = t.get("signal_type", "?") if st not in types: types[st] = {"count": 0, "pnl": 0, "wins": 0} types[st]["count"] += 1; types[st]["pnl"] += t.get("pnl", 0) if t.get("pnl", 0) > 0: types[st]["wins"] += 1 if types: print(f" By type :") for st, data in sorted(types.items()): wr = data["wins"]/data["count"]*100 if data["count"] else 0 print(f" {st:<16} {data['count']} trades P&L: ${data['pnl']:+.2f} WR: {wr:.0f}%") print(f"\n {'Time':>8} {'Type':>9} {'Side':>4} {'Entry':>7} {'Exit':>7} {'P&L':>7} Result") for t in exits[-30:]: print(f" {t.get('time_str','?'):>8} {t.get('signal_type','?'):>9} " f"{'YES' if t.get('side')=='yes' else 'NO ':>4} " f"${t.get('entry_price',0):.3f} ${t.get('exit_price',0):.3f} " f"${t.get('pnl',0):>+6.2f} {'WIN' if t.get('pnl',0)>0 else 'LOSS'}") print(f"\n Log: C:\\kalshibot\\paper_log.json") print("="*65) def run(): global balance, high_water_mark balance = STARTING_BALANCE; high_water_mark = STARTING_BALANCE # ── Reset paper balance to $100 on every restart (analytics kept separately) ── with open("C:\\kalshibot\\paper_log.json", "w") as f: json.dump({"balance": STARTING_BALANCE, "high_water_mark": STARTING_BALANCE, "trades": []}, f) print(f" ✅ Paper balance reset to ${STARTING_BALANCE:.2f} (trade_analytics.json preserved)") # Load auto-tuned params if the learning engine has generated them auto_params_path = "C:\\kalshibot\\auto_params.json" if os.path.exists(auto_params_path): try: with open(auto_params_path) as f: ap = json.load(f) print(f" ✅ Loaded auto_params.json ({ap.get('trades_analyzed','?')} trades analyzed)") for note in ap.get("notes", []): print(f" {note}") good_hours = ap.get("good_hours", []) avoid_hours = ap.get("avoid_hours", []) except Exception as e: print(f" ⚠ Could not load auto_params.json: {e}") good_hours = []; avoid_hours = [] else: good_hours = []; avoid_hours = [] print("\n" + "="*65) print(" PAPER TRADING BOT v3 -- $100 Starting Balance") print(" KXBTC15M | Binance (WS+REST) + Coinbase + Kraken") print("="*65) print(f" Entry modes : MOMENTUM ONLY (0.1%+ BTC move) — BALANCED disabled") print(f" Bet sizing : 2% of balance, max $2.00 per trade") print(f" Market lock : one trade per 15-min window (no double entries)") print(f" Position mon : every {FAST_MONITOR_INTERVAL}s (fast thread) | Market scan: every {SCAN_INTERVAL}s") print(f" R:R : +{PROFIT_TARGET*100:.0f}% target / -{STOP_LOSS*100:.0f}% stop = {PROFIT_TARGET/STOP_LOSS:.1f}:1") print(f" Trailing stop: halt if balance drops {TRAILING_STOP_PCT*100:.0f}% from PEAK") print(f" Dashboard : Double-click OPEN DASHBOARD on desktop") print("="*65) if WS_AVAILABLE: threading.Thread(target=binance_ws, daemon=True).start() threading.Thread(target=poll_binance_rest, daemon=True).start() threading.Thread(target=poll_coinbase, daemon=True).start() threading.Thread(target=poll_kraken, daemon=True).start() threading.Thread(target=fast_monitor_loop, daemon=True).start() # 2-second exit checker print(" All 4 price feeds + fast position monitor starting...") print(" Waiting for prices...") for i in range(30): time.sleep(1) with price_lock: has = any(p for p in prices.values() if p); ticks = len(price_history) if has and ticks >= WARMUP_TICKS: with price_lock: p = prices["binance"] or prices["coinbase"] or prices["kraken"] srcs = "+".join(k for k in prices if prices[k]) print(f" Ready BTC ${p:,.2f} [{srcs}]"); break elif has: with price_lock: p = prices["binance"] or prices["coinbase"] or prices["kraken"] print(f" BTC ${p:,.2f} ({ticks}/{WARMUP_TICKS} ticks)...") else: print(" Slow feeds — continuing anyway") print(f" Balance: ${balance:.2f} Stop at: ${balance*(1-TRAILING_STOP_PCT):.2f} | Ctrl+C to stop\n") last_entry = 0; current_tier = "BASE"; last_market_ticker = None while True: try: time.sleep(SCAN_INTERVAL); now = time.time() update_high_water_mark() btc, sources, edge = get_best_price() if not btc: print(f" [{datetime.datetime.now().strftime('%H:%M:%S')}] No price data"); continue momentum, direction, pct = get_momentum() market = get_15m_market() # Clear per-market lock when a new 15-min window opens if market and market["ticker"] != last_market_ticker: traded_tickers.clear() last_market_ticker = market["ticker"] print(f" [NEW WINDOW] {market['ticker'][-20:]} clearing market lock") with price_lock: ticks = len(price_history) warmup = ticks < WARMUP_TICKS should_trade, trade_dir, signal_type = get_signal(market, momentum, direction) if not warmup and momentum is not None: _, current_tier = get_bet_size(momentum, edge, signal_type or "MOMENTUM") if warmup: mom_str = f"warming up ({ticks}/{WARMUP_TICKS})" else: d = direction or "FLAT" sig_str = f"[{signal_type}]" if should_trade else "" mom_str = f"{'UP' if d=='UP' else 'DOWN' if d=='DOWN' else 'FLAT'} {pct} {sig_str}" stop_level = round(high_water_mark * (1 - TRAILING_STOP_PCT), 2) halt_str = " [HALTED]" if session_halted else "" mkt_str = "no market" if not market else \ f"{int(market['secs_left']//60)}m{int(market['secs_left']%60)}s " \ f"Y:{market['yes_ask']:.2f} N:{market['no_ask']:.2f}" print(f" [{datetime.datetime.now().strftime('%H:%M:%S')}] " f"BTC ${btc:,.2f} [{sources}] {mom_str} | {mkt_str} | " f"${balance:.2f} (peak:${high_water_mark:.2f} stop:${stop_level:.2f}) " f"[{len(paper_positions)}/{MAX_POSITIONS}]{halt_str}") if edge: print(f" {edge}") if paper_positions: print_position_status() save_state(btc, sources, edge, momentum, direction, market, current_tier) log_market_snapshot(btc, market, momentum, direction) # Auto-analyze every AUTO_RETUNE_EVERY completed trades completed = len([t for t in paper_log if t.get("event") == "exit"]) if completed > 0 and completed != _last_autotune_count and completed % AUTO_RETUNE_EVERY == 0: auto_analyze() current_hour = datetime.datetime.now().hour hour_blocked = (avoid_hours and current_hour in avoid_hours and not (good_hours and current_hour in good_hours)) can_trade = (not warmup and not check_session_stop_loss() and market is not None and should_trade and trade_dir is not None and len(paper_positions) < MAX_POSITIONS and now - last_entry > ENTRY_COOLDOWN and market["secs_left"] > MIN_SECS_LEFT and balance >= BET_MIN_ABS and not hour_blocked) if can_trade: entered, tier = paper_enter(market, trade_dir, signal_type, btc, momentum, edge) if entered: last_entry = now except KeyboardInterrupt: print_report(); save_log(); break except Exception as e: print(f" Error: {e}"); time.sleep(10) if __name__ == "__main__": run()