Add short signal scanner + leverage trading game engine + auto-trader
- short_scanner.py: RSI/VWAP/MACD/Bollinger-based short signal detection - leverage_game.py: Full game engine with longs/shorts/leverage/liquidations - leverage_trader.py: Auto-trader connecting scanners to game with TP/SL/trailing stops - Leverage Challenge game initialized: $10K, 20x max leverage, player 'case' - systemd timer: every 15min scan + trade - Telegram alerts on opens/closes/liquidations
This commit is contained in:
504
projects/crypto-signals/leverage_game.py
Normal file
504
projects/crypto-signals/leverage_game.py
Normal file
@ -0,0 +1,504 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Crypto Leverage Trading Game Engine
|
||||
Paper trading with longs, shorts, and configurable leverage.
|
||||
Tracks liquidation prices, unrealized PnL, and funding costs.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import uuid
|
||||
import time
|
||||
import urllib.request
|
||||
from datetime import datetime, date, timezone
|
||||
from pathlib import Path
|
||||
|
||||
DATA_DIR = Path(__file__).parent / "data" / "leverage-game"
|
||||
GAMES_DIR = DATA_DIR / "games"
|
||||
|
||||
BINANCE_TICKER = "https://api.binance.us/api/v3/ticker/price"
|
||||
|
||||
|
||||
def _load(path, default=None):
|
||||
if path.exists():
|
||||
return json.loads(path.read_text())
|
||||
return default if default is not None else {}
|
||||
|
||||
|
||||
def _save(path, data):
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(data, indent=2, default=str))
|
||||
|
||||
|
||||
def _game_path(game_id):
|
||||
return GAMES_DIR / game_id / "game.json"
|
||||
|
||||
def _player_path(game_id, username):
|
||||
return GAMES_DIR / game_id / "players" / username / "portfolio.json"
|
||||
|
||||
def _trades_path(game_id, username):
|
||||
return GAMES_DIR / game_id / "players" / username / "trades.json"
|
||||
|
||||
def _snapshots_path(game_id, username):
|
||||
return GAMES_DIR / game_id / "players" / username / "snapshots.json"
|
||||
|
||||
|
||||
# ── Price Fetching ──
|
||||
|
||||
def get_price(symbol):
|
||||
"""Get current price from Binance US."""
|
||||
if not symbol.endswith("USDT"):
|
||||
symbol = f"{symbol}USDT"
|
||||
url = f"{BINANCE_TICKER}?symbol={symbol}"
|
||||
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
|
||||
try:
|
||||
resp = urllib.request.urlopen(req, timeout=10)
|
||||
return float(json.loads(resp.read())['price'])
|
||||
except:
|
||||
return None
|
||||
|
||||
|
||||
# ── Game Management ──
|
||||
|
||||
def create_game(name, starting_cash=10_000.0, max_leverage=20, creator="system"):
|
||||
"""Create a new leverage trading game."""
|
||||
game_id = str(uuid.uuid4())[:8]
|
||||
config = {
|
||||
"game_id": game_id,
|
||||
"name": name,
|
||||
"starting_cash": starting_cash,
|
||||
"max_leverage": max_leverage,
|
||||
"funding_rate_8h": 0.01, # 0.01% per 8h (typical perp funding)
|
||||
"maker_fee": 0.02, # 0.02%
|
||||
"taker_fee": 0.05, # 0.05%
|
||||
"start_date": date.today().isoformat(),
|
||||
"creator": creator,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
"players": [],
|
||||
"status": "active",
|
||||
}
|
||||
_save(_game_path(game_id), config)
|
||||
return game_id
|
||||
|
||||
|
||||
def list_games(active_only=True):
|
||||
"""List all leverage games."""
|
||||
games = []
|
||||
if not GAMES_DIR.exists():
|
||||
return games
|
||||
for gid in os.listdir(GAMES_DIR):
|
||||
gp = _game_path(gid)
|
||||
if gp.exists():
|
||||
config = _load(gp)
|
||||
if active_only and config.get("status") != "active":
|
||||
continue
|
||||
games.append(config)
|
||||
return sorted(games, key=lambda g: g.get("created_at", ""), reverse=True)
|
||||
|
||||
|
||||
def get_game(game_id):
|
||||
return _load(_game_path(game_id))
|
||||
|
||||
|
||||
def join_game(game_id, username):
|
||||
"""Add player to game."""
|
||||
config = get_game(game_id)
|
||||
if not config:
|
||||
return {"error": "Game not found"}
|
||||
if username in config["players"]:
|
||||
return {"error": f"{username} already in game"}
|
||||
|
||||
config["players"].append(username)
|
||||
_save(_game_path(game_id), config)
|
||||
|
||||
_save(_player_path(game_id, username), {
|
||||
"cash": config["starting_cash"],
|
||||
"positions": {},
|
||||
"total_realized_pnl": 0,
|
||||
"total_fees_paid": 0,
|
||||
"total_funding_paid": 0,
|
||||
})
|
||||
_save(_trades_path(game_id, username), [])
|
||||
_save(_snapshots_path(game_id, username), [])
|
||||
|
||||
return {"success": True, "game_id": game_id, "username": username}
|
||||
|
||||
|
||||
# ── Position Math ──
|
||||
|
||||
def calc_liquidation_price(entry_price, leverage, direction):
|
||||
"""
|
||||
Simplified liquidation price.
|
||||
Long: liq = entry * (1 - 1/leverage)
|
||||
Short: liq = entry * (1 + 1/leverage)
|
||||
"""
|
||||
if direction == "long":
|
||||
return entry_price * (1 - 1 / leverage)
|
||||
else: # short
|
||||
return entry_price * (1 + 1 / leverage)
|
||||
|
||||
|
||||
def calc_unrealized_pnl(entry_price, current_price, size_usd, leverage, direction):
|
||||
"""
|
||||
Calculate unrealized PnL for a leveraged position.
|
||||
size_usd = margin (collateral). Notional = size_usd * leverage.
|
||||
"""
|
||||
notional = size_usd * leverage
|
||||
shares = notional / entry_price
|
||||
|
||||
if direction == "long":
|
||||
pnl = (current_price - entry_price) * shares
|
||||
else: # short
|
||||
pnl = (entry_price - current_price) * shares
|
||||
|
||||
return pnl
|
||||
|
||||
|
||||
def is_liquidated(entry_price, current_price, leverage, direction):
|
||||
"""Check if position would be liquidated."""
|
||||
liq_price = calc_liquidation_price(entry_price, leverage, direction)
|
||||
if direction == "long":
|
||||
return current_price <= liq_price
|
||||
else:
|
||||
return current_price >= liq_price
|
||||
|
||||
|
||||
# ── Trading ──
|
||||
|
||||
def open_position(game_id, username, symbol, direction, margin_usd, leverage, reason="Manual"):
|
||||
"""
|
||||
Open a leveraged position.
|
||||
margin_usd: collateral put up
|
||||
leverage: multiplier (e.g., 10x)
|
||||
direction: 'long' or 'short'
|
||||
"""
|
||||
pf = _load(_player_path(game_id, username))
|
||||
game = get_game(game_id)
|
||||
if not pf or not game:
|
||||
return {"error": "Player or game not found"}
|
||||
|
||||
if direction not in ("long", "short"):
|
||||
return {"error": "Direction must be 'long' or 'short'"}
|
||||
if leverage > game.get("max_leverage", 20):
|
||||
return {"error": f"Max leverage is {game['max_leverage']}x"}
|
||||
if margin_usd > pf["cash"]:
|
||||
return {"error": f"Insufficient cash. Need ${margin_usd:.2f}, have ${pf['cash']:.2f}"}
|
||||
|
||||
symbol = symbol.upper().replace("USDT", "")
|
||||
price = get_price(symbol)
|
||||
if not price:
|
||||
return {"error": f"Could not fetch price for {symbol}"}
|
||||
|
||||
notional = margin_usd * leverage
|
||||
fee = notional * game.get("taker_fee", 0.05) / 100
|
||||
|
||||
# Deduct margin + entry fee from cash
|
||||
pf["cash"] -= (margin_usd + fee)
|
||||
pf["total_fees_paid"] = pf.get("total_fees_paid", 0) + fee
|
||||
|
||||
pos_id = f"{symbol}_{direction}_{str(uuid.uuid4())[:4]}"
|
||||
liq_price = calc_liquidation_price(price, leverage, direction)
|
||||
|
||||
pf["positions"][pos_id] = {
|
||||
"symbol": symbol,
|
||||
"direction": direction,
|
||||
"leverage": leverage,
|
||||
"margin_usd": margin_usd,
|
||||
"notional": round(notional, 2),
|
||||
"entry_price": price,
|
||||
"current_price": price,
|
||||
"liquidation_price": round(liq_price, 4),
|
||||
"unrealized_pnl": 0,
|
||||
"entry_fee": round(fee, 4),
|
||||
"opened_at": datetime.now(timezone.utc).isoformat(),
|
||||
"reason": reason,
|
||||
}
|
||||
|
||||
_save(_player_path(game_id, username), pf)
|
||||
|
||||
# Log trade
|
||||
trades = _load(_trades_path(game_id, username), [])
|
||||
trades.append({
|
||||
"action": "OPEN",
|
||||
"pos_id": pos_id,
|
||||
"symbol": symbol,
|
||||
"direction": direction,
|
||||
"leverage": leverage,
|
||||
"margin_usd": margin_usd,
|
||||
"notional": round(notional, 2),
|
||||
"entry_price": price,
|
||||
"liquidation_price": round(liq_price, 4),
|
||||
"fee": round(fee, 4),
|
||||
"reason": reason,
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
})
|
||||
_save(_trades_path(game_id, username), trades)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"pos_id": pos_id,
|
||||
"symbol": symbol,
|
||||
"direction": direction,
|
||||
"leverage": leverage,
|
||||
"entry_price": price,
|
||||
"margin": margin_usd,
|
||||
"notional": round(notional, 2),
|
||||
"liquidation_price": round(liq_price, 4),
|
||||
"fee": round(fee, 4),
|
||||
}
|
||||
|
||||
|
||||
def close_position(game_id, username, pos_id, reason="Manual"):
|
||||
"""Close a leveraged position."""
|
||||
pf = _load(_player_path(game_id, username))
|
||||
game = get_game(game_id)
|
||||
if not pf or not game:
|
||||
return {"error": "Player or game not found"}
|
||||
if pos_id not in pf["positions"]:
|
||||
return {"error": f"Position {pos_id} not found"}
|
||||
|
||||
pos = pf["positions"][pos_id]
|
||||
price = get_price(pos["symbol"])
|
||||
if not price:
|
||||
return {"error": f"Could not fetch price for {pos['symbol']}"}
|
||||
|
||||
# Calculate PnL
|
||||
pnl = calc_unrealized_pnl(
|
||||
pos["entry_price"], price, pos["margin_usd"], pos["leverage"], pos["direction"]
|
||||
)
|
||||
|
||||
# Check liquidation
|
||||
liquidated = is_liquidated(pos["entry_price"], price, pos["leverage"], pos["direction"])
|
||||
if liquidated:
|
||||
pnl = -pos["margin_usd"] # Lose entire margin
|
||||
|
||||
# Exit fee
|
||||
notional = pos["margin_usd"] * pos["leverage"]
|
||||
fee = notional * game.get("taker_fee", 0.05) / 100
|
||||
|
||||
# Return margin + PnL - fee to cash
|
||||
returned = pos["margin_usd"] + pnl - fee
|
||||
if returned < 0:
|
||||
returned = 0 # Can't lose more than margin (no negative balance)
|
||||
|
||||
pf["cash"] += returned
|
||||
pf["total_realized_pnl"] = pf.get("total_realized_pnl", 0) + pnl
|
||||
pf["total_fees_paid"] = pf.get("total_fees_paid", 0) + fee
|
||||
|
||||
del pf["positions"][pos_id]
|
||||
_save(_player_path(game_id, username), pf)
|
||||
|
||||
# Log trade
|
||||
pnl_pct = (pnl / pos["margin_usd"] * 100) if pos["margin_usd"] > 0 else 0
|
||||
trades = _load(_trades_path(game_id, username), [])
|
||||
trades.append({
|
||||
"action": "LIQUIDATED" if liquidated else "CLOSE",
|
||||
"pos_id": pos_id,
|
||||
"symbol": pos["symbol"],
|
||||
"direction": pos["direction"],
|
||||
"leverage": pos["leverage"],
|
||||
"entry_price": pos["entry_price"],
|
||||
"exit_price": price,
|
||||
"margin_usd": pos["margin_usd"],
|
||||
"pnl": round(pnl, 2),
|
||||
"pnl_pct": round(pnl_pct, 2),
|
||||
"fee": round(fee, 4),
|
||||
"liquidated": liquidated,
|
||||
"reason": reason,
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
})
|
||||
_save(_trades_path(game_id, username), trades)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"pos_id": pos_id,
|
||||
"symbol": pos["symbol"],
|
||||
"direction": pos["direction"],
|
||||
"entry_price": pos["entry_price"],
|
||||
"exit_price": price,
|
||||
"pnl": round(pnl, 2),
|
||||
"pnl_pct": round(pnl_pct, 2),
|
||||
"liquidated": liquidated,
|
||||
"returned_to_cash": round(returned, 2),
|
||||
}
|
||||
|
||||
|
||||
def update_prices(game_id, username):
|
||||
"""Update all position prices and check for liquidations."""
|
||||
pf = _load(_player_path(game_id, username))
|
||||
if not pf:
|
||||
return []
|
||||
|
||||
liquidations = []
|
||||
to_liquidate = []
|
||||
|
||||
for pos_id, pos in pf["positions"].items():
|
||||
price = get_price(pos["symbol"])
|
||||
if not price:
|
||||
continue
|
||||
|
||||
pos["current_price"] = price
|
||||
pos["unrealized_pnl"] = round(
|
||||
calc_unrealized_pnl(pos["entry_price"], price, pos["margin_usd"], pos["leverage"], pos["direction"]),
|
||||
2
|
||||
)
|
||||
|
||||
if is_liquidated(pos["entry_price"], price, pos["leverage"], pos["direction"]):
|
||||
to_liquidate.append(pos_id)
|
||||
|
||||
time.sleep(0.1)
|
||||
|
||||
_save(_player_path(game_id, username), pf)
|
||||
|
||||
# Process liquidations
|
||||
for pos_id in to_liquidate:
|
||||
result = close_position(game_id, username, pos_id, reason="LIQUIDATED")
|
||||
liquidations.append(result)
|
||||
|
||||
return liquidations
|
||||
|
||||
|
||||
# ── Portfolio View ──
|
||||
|
||||
def get_portfolio(game_id, username):
|
||||
"""Get full portfolio with live PnL."""
|
||||
pf = _load(_player_path(game_id, username))
|
||||
game = get_game(game_id)
|
||||
if not pf or not game:
|
||||
return None
|
||||
|
||||
starting = game["starting_cash"]
|
||||
total_unrealized = sum(p.get("unrealized_pnl", 0) for p in pf["positions"].values())
|
||||
total_margin_locked = sum(p["margin_usd"] for p in pf["positions"].values())
|
||||
equity = pf["cash"] + total_margin_locked + total_unrealized
|
||||
total_pnl = equity - starting
|
||||
|
||||
return {
|
||||
"username": username,
|
||||
"game_id": game_id,
|
||||
"cash": round(pf["cash"], 2),
|
||||
"margin_locked": round(total_margin_locked, 2),
|
||||
"unrealized_pnl": round(total_unrealized, 2),
|
||||
"realized_pnl": round(pf.get("total_realized_pnl", 0), 2),
|
||||
"total_fees": round(pf.get("total_fees_paid", 0), 2),
|
||||
"equity": round(equity, 2),
|
||||
"total_pnl": round(total_pnl, 2),
|
||||
"pnl_pct": round(total_pnl / starting * 100, 2),
|
||||
"num_positions": len(pf["positions"]),
|
||||
"positions": pf["positions"],
|
||||
}
|
||||
|
||||
|
||||
def get_trades(game_id, username):
|
||||
return _load(_trades_path(game_id, username), [])
|
||||
|
||||
|
||||
def daily_snapshot(game_id, username):
|
||||
"""Take daily snapshot."""
|
||||
p = get_portfolio(game_id, username)
|
||||
if not p:
|
||||
return None
|
||||
snapshots = _load(_snapshots_path(game_id, username), [])
|
||||
today = date.today().isoformat()
|
||||
snapshots = [s for s in snapshots if s["date"] != today]
|
||||
snapshots.append({
|
||||
"date": today,
|
||||
"equity": p["equity"],
|
||||
"total_pnl": p["total_pnl"],
|
||||
"pnl_pct": p["pnl_pct"],
|
||||
"cash": p["cash"],
|
||||
"num_positions": p["num_positions"],
|
||||
"realized_pnl": p["realized_pnl"],
|
||||
"total_fees": p["total_fees"],
|
||||
})
|
||||
_save(_snapshots_path(game_id, username), snapshots)
|
||||
return snapshots[-1]
|
||||
|
||||
|
||||
def get_leaderboard(game_id):
|
||||
"""Leaderboard sorted by equity."""
|
||||
game = get_game(game_id)
|
||||
if not game:
|
||||
return []
|
||||
board = []
|
||||
for username in game["players"]:
|
||||
p = get_portfolio(game_id, username)
|
||||
if p:
|
||||
trades = get_trades(game_id, username)
|
||||
closed = [t for t in trades if t.get("action") in ("CLOSE", "LIQUIDATED")]
|
||||
wins = [t for t in closed if t.get("pnl", 0) > 0]
|
||||
liquidations = [t for t in closed if t.get("liquidated")]
|
||||
board.append({
|
||||
"username": username,
|
||||
"equity": p["equity"],
|
||||
"total_pnl": p["total_pnl"],
|
||||
"pnl_pct": p["pnl_pct"],
|
||||
"num_positions": p["num_positions"],
|
||||
"trades_closed": len(closed),
|
||||
"win_rate": round(len(wins) / len(closed) * 100, 1) if closed else 0,
|
||||
"liquidations": len(liquidations),
|
||||
"total_fees": p["total_fees"],
|
||||
})
|
||||
return sorted(board, key=lambda x: x["pnl_pct"], reverse=True)
|
||||
|
||||
|
||||
# ── Auto-Trader (Scanner Integration) ──
|
||||
|
||||
def auto_trade_from_scanner(game_id, username, scan_results, margin_per_trade=200, leverage=10):
|
||||
"""
|
||||
Automatically open positions based on scanner results.
|
||||
Short scanner (score >= 50) → open short
|
||||
Spot scanner (score >= 40) → open long
|
||||
"""
|
||||
opened = []
|
||||
for r in scan_results:
|
||||
symbol = r["symbol"]
|
||||
score = r["score"]
|
||||
|
||||
# Determine direction based on which scanner produced this
|
||||
direction = r.get("direction", "short") # Default to short for short scanner
|
||||
|
||||
if score < 40:
|
||||
continue
|
||||
|
||||
# Scale leverage with conviction
|
||||
if score >= 70:
|
||||
lev = min(leverage, 15)
|
||||
elif score >= 50:
|
||||
lev = min(leverage, 10)
|
||||
else:
|
||||
lev = min(leverage, 5)
|
||||
|
||||
result = open_position(game_id, username, symbol, direction, margin_per_trade, lev,
|
||||
reason=f"Scanner score:{score} | {', '.join(r.get('reasons', []))}")
|
||||
if result.get("success"):
|
||||
opened.append(result)
|
||||
|
||||
return opened
|
||||
|
||||
|
||||
# ── Initialize ──
|
||||
|
||||
def ensure_default_game():
|
||||
"""Create default Leverage Challenge game."""
|
||||
for g in list_games():
|
||||
if g["name"] == "Leverage Challenge":
|
||||
return g["game_id"]
|
||||
|
||||
game_id = create_game("Leverage Challenge", starting_cash=10_000.0, max_leverage=20, creator="case")
|
||||
join_game(game_id, "case")
|
||||
return game_id
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
game_id = ensure_default_game()
|
||||
game = get_game(game_id)
|
||||
print(f"Game: {game['name']} ({game_id})")
|
||||
print(f"Players: {game['players']}")
|
||||
print(f"Starting cash: ${game['starting_cash']:,.2f}")
|
||||
print(f"Max leverage: {game['max_leverage']}x")
|
||||
|
||||
board = get_leaderboard(game_id)
|
||||
for entry in board:
|
||||
print(f" {entry['username']}: ${entry['equity']:,.2f} ({entry['pnl_pct']:+.2f}%) "
|
||||
f"| {entry['trades_closed']} trades | {entry['win_rate']}% win | {entry['liquidations']} liquidated")
|
||||
Reference in New Issue
Block a user