# Jev Starter: build your own AI trading bot > **How to use this:** open Claude Code in an empty folder, drag this file in (or paste it), and say "go". > Claude sets everything up and talks you through it. Mac, Windows and Linux. > - **Path 1, laptop (about 15 min):** a live AI trading dashboard, a news terminal, and a strategy file you edit. > Paper trading: live prices, simulated trades, no exchange account. > - **Path 2, 24/7 bot (about 45–60 min more):** the same bot on an always-on server, with Claude as the brain, > trading on Bybit. It starts on Bybit's **demo account (fake money)**. --- You are the setup agent for **jev-starter**. Do the work yourself: run every command, check every result, fix errors as they come up. The user is watching and may not be technical, so talk to them in plain English, one or two short lines per step. Ask before anything that costs money or creates an account; those are their decisions. ## What you're building (say this to the user first, in 3–4 lines) Jev is a fast AI that answers multiple-choice questions ("buy or sell?") with a confidence score in about half a second, for a fraction of a cent. This builds a live trading dashboard where Jev makes calls on a real market several times a second, a news terminal where Jev reads real crypto headlines, and a strategy file that decides which calls become trades. Optionally, it then moves the bot to a 24/7 server with Claude as the big-picture brain, trading on Bybit. Then ask: **"Do you want just the laptop version (Path 1), or the full 24/7 bot too (Path 2)? Either way we start with the laptop version, so you can see it working first."** ## Rules (never break these) 1. Write every file exactly as given in the **Files** section. Don't rewrite, reformat or "improve" them. The only file you may change later is `jevlab/strategy.py`, when the user asks for a strategy. 2. Never type, paste, print or echo any key, secret or password (Vercel, Bybit, server). The user pastes keys into `.env` themselves. Never `cat` the `.env` file. Copying the file to their own server is fine. 3. Never set `BYBIT_MODE=live` or `JEV_LIVE_CONFIRM` yourself, never enable withdrawal permissions on an API key, and never weaken the safety checks in `bybit.py` / `bot.py`. Real money is the user's own deliberate step (Step B10). 4. Never open the dashboard port (8765) to the internet. On the server it's reached only through an SSH tunnel. 5. Never promise profit. The honest answer: it depends on the strategy, costs matter, and it needs weeks of demo results before any real money. Not financial advice. # PART A · Path 1: the laptop version (everyone does this) ## A1 · Check the machine 1. Work out the OS (`uname -s`; if that fails, it's Windows). 2. Run `uv --version`. If it's missing, install it: - Mac/Linux: `curl -LsSf https://astral.sh/uv/install.sh | sh`, then `export PATH="$HOME/.local/bin:$PATH"` - Windows (PowerShell): `powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"`, then restart the terminal if `uv` still isn't found 3. `uv` downloads its own Python, so nothing else needs installing. ## A2 · Write the project Create a folder called `jev-starter`, write every file from the **Files** section into it at the exact path shown, then `cd jev-starter` and run `uv sync`. ## A3 · Test it before the key Run `uv run python -m jevlab check`. Prices and news should say ok, and Jev will say "no key yet". That's expected. ## A4 · The Jev key (the user does this part) 1. Copy `.env.example` to `.env` (`cp .env.example .env`, or `copy .env.example .env` on Windows). 2. Open https://vercel.com/dashboard for them (`open` on Mac, `start` on Windows, `xdg-open` on Linux) and explain: sign in, go to **AI Gateway → API Keys → Create key**, and copy it. The free tier works but is slow. **$5 of AI Gateway credits** makes Jev full speed, and **Path 2 needs credits** because Claude runs through the same key. Buying credits is their choice; never do it for them. 3. Open `.env` in their editor (`open -e .env` on Mac, `notepad .env` on Windows, `${EDITOR:-nano} .env` on Linux) and say: *"Paste your key right after `AI_GATEWAY_API_KEY=`, save, and tell me when you're done."* Then wait. 4. Run `uv run python -m jevlab check` again. It should show Jev's answer and speed, e.g. "BUY 72% in 480 ms". "key rejected" → the key was pasted wrong or the Vercel team needs a card or credits. "rate-limited" → wait a minute, or add credits. ## A5 · The live dashboard Run `uv run python -m jevlab loop` **in the background** (it runs for 10 minutes, and the dashboard stays open until stopped). It opens http://127.0.0.1:8765. Walk them through it in plain English: - the big number: the live price, straight from the exchange - each dot on the chart: a Jev call (green = buy, red = sell), with confidence and response time in the feed - ◆ markers: actual (paper) trades. Most calls end in "hold": every trade has a cost, so the strategy is picky - the right-hand panel: the paper balance, before and after fees If the port is busy, add `--port 8766`. `--coin BTC` watches Bitcoin instead. ## A6 · The newsroom When the loop is done (or stopped), run `uv run python -m jevlab newsroom` in the background. It replays the last 36 hours of real crypto headlines in fast motion, round after round. Jev reads each one live (which coin, good or bad news, how big), and the chart shows what the price actually did next. Tell them it's a fast replay of real headlines, because live news is too slow to watch. Stop it when they've seen enough. ## A7 · Plug in their own strategy Open `jevlab/strategy.py` and explain it in two lines: every time Jev makes a call, `decide()` chooses buy, sell, flat or hold, and the default is "only act when Jev is 85%+ sure, and wait 15 seconds between flips". Ask: *"Do you have a strategy idea? Describe it in one sentence and I'll plug it in."* If they give one, rewrite `decide()` (plus `SETTINGS` / `DESCRIPTION`) to match, using only the inputs in its docstring and the same return values, then re-run `uv run python -m jevlab loop`. If not, show the trend-following example at the bottom of the file. Before trusting any strategy, it should pass three questions: why should it make money, does it survive costs, does it work on data it's never seen. **If they only wanted Path 1, wrap up here:** remind them of `check`, `loop` (`--minutes 0` = until Ctrl+C), `newsroom` and `strategy.py`, and that this is paper trading. # PART B · Path 2: the 24/7 bot (Claude + Jev on Bybit) Tell them what's coming in 3 lines: a Bybit demo account (fake money) and API key, a small always-on server from Hostinger, and the bot running there 24/7. Claude reads the market every 10 minutes and sets the direction, Jev makes the fast calls, and hard limits are always on. They need AI Gateway credits for Claude (Step A4). ## B1 · Bybit account and demo key (the user does the clicking) 1. **Check they can use Bybit:** Bybit isn't available in every country (for example the US), and they should check Bybit's terms for where they live. If they can't use it, stop Path 2 here and explain why. 2. If they don't have an account, open Miles's Bybit referral link for them: https://partner.bybit.com/b/166453 (say it's Miles's referral link, with welcome rewards for new users). They sign up and complete any verification Bybit asks for. 3. In Bybit, switch to **Demo Trading** (in the account menu). Demo has its own fake balance; top it up there if it's empty. 4. While in Demo Trading, create an API key: **API → Create New Key → system-generated**, with **Read-Write** permission for **Contracts / Derivatives: Orders and Positions** only. **No withdrawals, no transfers.** Leave the IP restriction for now; we lock it to the server in B5. Labels on Bybit's site can change a little; guide them by meaning. 5. Open `.env` in their editor and have them paste the key after `BYBIT_API_KEY=` and the secret after `BYBIT_API_SECRET=`, and save. `BYBIT_MODE=demo` is already set. 6. Run `uv run python -m jevlab check`. Step 5 should say "connected to Bybit DEMO" with the demo balance, and "key can't withdraw". If it says the key can withdraw, have them delete it and make a new one without that permission. ## B2 · Test the bot on the laptop first Run `uv run python -m jevlab bot --minutes 5` in the background. The dashboard opens with Claude's bias box (e.g. "Claude's bias: FLAT · reason…"), Jev's calls, and the bot's limit orders on Bybit's demo account. Ask them to open Bybit Demo Trading → orders / positions to see the same orders appear. Explain that "hold · Claude says stay out" is normal: in quiet or choppy markets Claude often sits out, and then there are no orders at all. That's the brain doing its job, not a bug. When Claude picks a side, the orders appear. Stop it after 5 minutes. (`uv run python -m jevlab bot --dry` does the same on Bybit prices with simulated fills, with no Bybit key.) ## B3 · Get a server (the user does the buying) 1. Open Miles's Hostinger link for them: https://www.hostinger.com/aiedge (Miles's referral link; the code `aiedge` gives an extra discount). They choose a **VPS** plan (the smallest KVM plan is plenty), with **Ubuntu 24.04**. 2. **Pick a data centre outside the US** (Bybit blocks US server addresses), in a country where they're allowed to use Bybit. 3. Before they finish setup, make an SSH key for them. It's a key pair: the private half stays on their laptop, and the public half goes to the server. - Mac/Linux: `ssh-keygen -t ed25519 -f ~/.ssh/jev_vps -N "" -C jev-vps`, then show `cat ~/.ssh/jev_vps.pub` - Windows: `ssh-keygen -t ed25519 -f "$env:USERPROFILE\.ssh\jev_vps" -N '""' -C jev-vps`, then show `Get-Content "$env:USERPROFILE\.ssh\jev_vps.pub"` Showing the **.pub** (public) key is safe. Tell them to paste it into Hostinger's **SSH key** field. If Hostinger asks for a root password, they choose one themselves; you never need it. 4. When the server is ready, ask for its **IP address** (shown in Hostinger's VPS dashboard). ## B4 · Install the bot on the server Below, `$SSH` is shorthand for the full command `ssh -i ~/.ssh/jev_vps -o StrictHostKeyChecking=accept-new root@` (on Windows, use the `$env:USERPROFILE\.ssh\jev_vps` key path). Each command runs in a fresh shell, so write the full command every time rather than relying on a variable. 1. Test it: `$SSH "echo connected"`. 2. Install uv there: `$SSH "curl -LsSf https://astral.sh/uv/install.sh | sh"`. 3. Copy the project, including `.env` (as a file; never print it), but not `.venv` or `results`: - Mac/Linux: `rsync -az --exclude .venv --exclude results -e "ssh -i ~/.ssh/jev_vps" ./ root@:/root/jev-starter/` - Windows (no rsync): `$SSH "mkdir -p /root/jev-starter"`, then `scp -i -r pyproject.toml README.md .env jevlab dashboard deploy root@:/root/jev-starter/` 4. `$SSH "cd /root/jev-starter && /root/.local/bin/uv sync && /root/.local/bin/uv run python -m jevlab check"`: Jev and Bybit DEMO should both say ok **from the server**. That also confirms the server's location isn't blocked. ## B5 · Lock the Bybit key to the server Have them edit the demo API key on Bybit and restrict it to the server's IP address only. Re-run the server check from B4.4, which should now say "key is locked to IP". ## B6 · Run it 24/7 `$SSH "cp /root/jev-starter/deploy/jev-bot.service /etc/systemd/system/ && systemctl daemon-reload && systemctl enable --now jev-bot"` Then `$SSH "systemctl status jev-bot --no-pager"` and `$SSH "journalctl -u jev-bot -n 30 --no-pager"` should show it running. It restarts itself if it crashes or the server reboots. ## B7 · Watch it from the laptop Open a private tunnel in the background: `ssh -i ~/.ssh/jev_vps -N -L 8765:127.0.0.1:8765 root@`, then open http://127.0.0.1:8765. It's the same dashboard, now showing the bot on the server. (If port 8765 is busy on the laptop, use `-L 8766:127.0.0.1:8765` and open :8766.) The dashboard is never exposed to the internet. ## B8 · Controls (tell them, and write these into a short note for them) - **Stop:** `$SSH "systemctl stop jev-bot"` (it cancels orders and closes the position). Start again: `systemctl start jev-bot` - **Kill switch:** `$SSH "touch /root/jev-starter/STOP"` closes out and halts. Delete the file and restart to resume. - **Logs:** `$SSH "journalctl -u jev-bot -f"` - **Change the strategy:** edit `jevlab/strategy.py` on the laptop, test with `bot --minutes 5`, copy it up (re-run the B4.3 copy), then `$SSH "systemctl restart jev-bot"` - **Limits:** `MAX_POSITION_USD` and `MAX_DAILY_LOSS_USD` in the server's `.env`. The bot halts for the day at the loss limit. ## B9 · Wrap up Tell them it's now running 24/7 on Bybit **demo**, with Claude setting the direction and Jev making the fast calls. Suggest they leave it on demo for at least a few weeks and judge it on after-fee results before even thinking about real money. ## B10 · Real money (only if they explicitly ask) Explain the steps, but they do them themselves: create a **live** API key with the same rules (orders and positions only, no withdrawals, locked to the server IP), put it in the server's `.env`, set small limits first (e.g. `MAX_POSITION_USD=50`, `MAX_DAILY_LOSS_USD=10`), and set `BYBIT_MODE=live` and `JEV_LIVE_CONFIRM=I accept the risk of trading real money` themselves. Then restart the service. Remind them: only risk money they can afford to lose. Not financial advice. --- ## Files ### `jev-starter/pyproject.toml` ````toml [project] name = "jev-starter" version = "1.0.0" description = "A Jev trading bot starter: live dashboard, news terminal, a strategy file you edit, and an optional 24/7 Bybit bot (demo first)." requires-python = ">=3.10" dependencies = [ "requests>=2.31", "pandas>=2.0", "python-dotenv>=1.0", "rich>=13", "websocket-client>=1.7", "pybit>=5.8", ] ```` ### `jev-starter/.gitignore` ```` .env .venv/ results/ __pycache__/ ```` ### `jev-starter/.env.example` ```` # Copy this file to .env (same folder) and paste your keys after the = signs. # Never share this file. # --- Jev (and Claude) ------------------------------------------------------- # Vercel AI Gateway key: https://vercel.com/dashboard -> AI Gateway -> API Keys -> Create key # The free tier works but is slow (about one Jev call every 2 seconds). # $5 of AI Gateway credits removes that limit, and is needed for Claude (the 24/7 bot's brain). AI_GATEWAY_API_KEY= # Optional: a direct TypeSafe key, if you're off their waitlist. TYPESAFE_API_KEY= # --- 24/7 bot only (Path 2) ------------------------------------------------- # Bybit API key and secret. Start with a DEMO TRADING key (fake money). # The key must NOT have withdrawal permission. Lock it to your server's IP. BYBIT_API_KEY= BYBIT_API_SECRET= BYBIT_MODE=demo # Hard limits, always on: MAX_POSITION_USD=1000 MAX_DAILY_LOSS_USD=50 # Claude model for the big-picture brain (via the same Vercel key). CLAUDE_MODEL=anthropic/claude-sonnet-5 # Real money: leave this blank. Only set BYBIT_MODE=live, and this line to the exact # phrase "I accept the risk of trading real money", after weeks of demo results. JEV_LIVE_CONFIRM= ```` ### `jev-starter/README.md` ````markdown # jev-starter A Jev trading bot you can watch, test, and plug your own strategy into. **The laptop version is paper trading only**: it reads live public market data and simulates fills, with no exchange account needed. The optional **24/7 bot** (Path 2) trades on Bybit, starting on Bybit Demo Trading (fake money). ## What's inside | | | |---|---| | **Jev Loop** · `uv run python -m jevlab loop` | Live dashboard: the real price ticking about 8 times a second, Jev's BUY/SELL calls landing on it, and your strategy deciding which calls become trades. Shows P&L before and after fees on a $1,000 paper position. | | **Newsroom** · `uv run python -m jevlab newsroom` | A news terminal. It replays the last 36 hours of real crypto headlines in fast motion, Jev reads each one live (which coin, which way, how big), and then shows what the price actually did next. | | **Your strategy** · `jevlab/strategy.py` | The one file you edit. `decide()` gets Jev's call and the live market numbers, and returns `"buy"`, `"sell"` or `"hold · reason"`. | | **Setup check** · `uv run python -m jevlab check` | Tests prices, news feeds, your Jev key and your speed tier. | ## Setup 1. Install [uv](https://docs.astral.sh/uv/). 2. `cp .env.example .env` and paste your Vercel AI Gateway key after `AI_GATEWAY_API_KEY=`. Get one at vercel.com/dashboard → AI Gateway → API Keys. The free tier works but is slow (about one Jev call every 2 seconds); $5 of AI Gateway credits runs it at full speed. Each call costs a tiny fraction of a cent. 3. `uv sync`, then `uv run python -m jevlab check`. ## Plug in your own strategy Open `jevlab/strategy.py`. The default is three simple rules: - only act when Jev is at least 85% sure - wait 15 seconds between flips - let the loop use limit orders Change `SETTINGS`, or rewrite `decide()`. There's a trend-following example at the bottom of the file. Easiest route: tell Claude your idea in one sentence and ask it to rewrite `decide()`. Then run `uv run python -m jevlab loop` again and watch what changes. Before trusting any strategy, ask three questions: 1. Why should it make money? 2. Does it still make money after costs? 3. Does it work on data it's never seen? ## Flags `--coin BTC` · `--minutes 0` (run until Ctrl+C) · `--taker` (market orders) · `--gap 4` (newsroom pace) · `--port 8766` (a second dashboard) · `--no-open` ## The 24/7 bot (Path 2): Claude + Jev on Bybit `uv run python -m jevlab bot` runs the same loop against Bybit, with Claude as the big-picture brain: - **Claude** reads the market every 10 minutes and sets the bias: long, short or flat - **Jev** makes the fast calls - **`strategy.py`** only trades in Claude's direction It starts on **Bybit Demo Trading** (fake money). `--dry` runs it on Bybit prices with simulated fills and needs no Bybit key. Safety, always on: - post-only limit orders - `MAX_POSITION_USD` and `MAX_DAILY_LOSS_USD` limits - a kill switch: create a file called `STOP` and it closes out and halts - on shutdown it cancels orders and closes the position Real money needs `BYBIT_MODE=live` **and** the exact confirmation phrase in `JEV_LIVE_CONFIRM`. Set those yourself, after weeks of demo results. To run it 24/7, put it on an always-on server (a small Linux VPS) with `deploy/jev-bot.service`. The setup prompt walks through all of it. Not financial advice. ```` ### `jev-starter/jevlab/__init__.py` ````python """jev-starter: a Jev trading bot you can see, test and plug your own strategy into. Paper only.""" ```` ### `jev-starter/jevlab/__main__.py` ````python """jev-starter CLI. uv run python -m jevlab check # test the setup: live prices, headlines, and one real Jev call uv run python -m jevlab loop # the Jev Loop: live trading dashboard (paper), your strategy uv run python -m jevlab newsroom # the news terminal: real headlines, Jev reading each one live uv run python -m jevlab bot --dry # 24/7 bot on Bybit prices with simulated fills (no Bybit key) uv run python -m jevlab bot # 24/7 bot on Bybit Demo Trading (fake money) · Claude sets the bias Useful flags: --coin BTC loop/bot: which coin to trade (default HYPE) --minutes 0 loop: run until Ctrl+C (default 10) --taker loop: market orders instead of limit orders --gap 4 newsroom: seconds between replayed headlines (default 6) --port 8766 run a second dashboard alongside the first --no-open don't open the browser """ from __future__ import annotations import argparse import threading def check() -> None: import time from . import hl from .core import console from .judges import JevJudge, JudgeError from .news import fetch_headlines console.print(" [bold]1. Hyperliquid prices[/]") mids = hl.mids(["BTC", "HYPE"]) console.print(f" BTC {mids['BTC']:,.1f} · HYPE {mids['HYPE']:,.3f} [#3fd68a]ok[/]") console.print(" [bold]2. News feeds[/]") n = len(fetch_headlines(24)) console.print(f" {n} headlines in the last 24h [#3fd68a]ok[/]" if n else " no headlines found [#f5b53d]check your internet[/]") console.print(" [bold]3. Jev[/]") try: jev = JevJudge() except JudgeError: console.print(" [#f5b53d]no key yet[/]: paste your AI_GATEWAY_API_KEY into .env, save, and run this again") return q = {"side": {"type": "choice", "instructions": "Which side should a bot hold for the next few seconds?", "criteria": {"buy": None, "sell": None}}} try: ans, meta = jev.ask({"return_5s_bps": 1.2, "top_of_book_imbalance": 0.4}, q, timeout=15, retries=2) except JudgeError as exc: msg = str(exc) if "401" in msg or "403" in msg: console.print(" [#ff5d6c]key rejected[/]: check the key in .env, and that your Vercel team has a card or credits") elif "429" in msg: console.print(" [#f5b53d]rate-limited[/]: the free tier is busy. Wait a minute, or add $5 of AI Gateway credits for full speed") else: console.print(f" [#ff5d6c]failed[/]: {msg[:160]}") return side = ans["side"]["choice"] console.print(f" Jev says [bold]{side.upper()} {ans['side']['probs'][side]:.0%}[/] in {meta['latency_ms']} ms " f"(model {meta['model']}) [#3fd68a]ok[/]") console.print(" [bold]4. Speed[/]") ok = 0 t0 = time.time() for _ in range(6): try: jev.ask({"return_5s_bps": 0.5}, q, timeout=10, retries=0) ok += 1 except JudgeError: pass time.sleep(0.3) console.print(f" {ok}/6 quick calls went through in {time.time() - t0:.1f}s · " + ("full speed" if ok == 6 else "you're on the free tier, so the loop will run slower (add $5 of credits for full speed)")) check_bybit() console.print("\n [#3fd68a]All set.[/] Next: [bold]uv run python -m jevlab loop[/]") def check_bybit() -> None: import os from .core import console console.print(" [bold]5. Bybit (only needed for the 24/7 bot)[/]") if not os.getenv("BYBIT_API_KEY", "").strip(): console.print(" not set up yet (fine for the laptop version)") return from .bybit import BybitError, BybitTrader, bybit_mode try: mode = bybit_mode() trader = BybitTrader(os.getenv("BOT_SYMBOL", "HYPEUSDT"), mode) eq = trader.equity_usdt() console.print(f" connected to Bybit [bold]{mode.upper()}[/] · equity {eq:,.2f} USDT [#3fd68a]ok[/]" if eq is not None else f" connected to Bybit {mode.upper()} [#3fd68a]ok[/]") except BybitError as exc: msg = str(exc) hint = (" (a demo key only works with BYBIT_MODE=demo, a live key only with live)" if "10003" in msg or "API key is invalid" in msg else " (check the key and secret in .env)") console.print(f" [#ff5d6c]Bybit said no[/]: {msg[:140]}{hint}") return try: # safety checks on the key itself (not every account type exposes this) info = trader._call(trader.http.get_api_key_information) perms = info.get("permissions", {}) or {} flat = {p for v in perms.values() for p in (v or [])} if any("Withdraw" in p for p in flat): console.print(" [#ff5d6c]this key can WITHDRAW funds[/]: make a new key without withdrawal permission") else: console.print(" key can't withdraw [#3fd68a]ok[/]") ips = info.get("ips") or [] console.print(" key is locked to IP " + ", ".join(ips) + " [#3fd68a]ok[/]" if ips and ips != ["*"] else " key isn't locked to an IP yet (lock it to your server's IP once the server is set up)") except Exception: console.print(" [dim](couldn't read the key's permissions here, so double-check them on Bybit: no withdrawals)[/]") def main() -> None: ap = argparse.ArgumentParser(prog="jevlab") ap.add_argument("command", choices=["check", "loop", "newsroom", "bot"]) ap.add_argument("--coin", default="HYPE", help="loop/bot: the coin to trade") ap.add_argument("--minutes", type=float, default=10.0, help="loop/bot: how long to run (0 = until stopped)") ap.add_argument("--pace", type=float, default=0.3, help="loop: fastest seconds between Jev calls") ap.add_argument("--late-ms", type=float, default=1500, help="loop: answers slower than this are ignored") ap.add_argument("--taker", action="store_true", help="loop: market orders instead of limit orders") ap.add_argument("--maker-wait", type=float, default=10.0, help="loop: seconds a limit order rests before cancel") ap.add_argument("--hours", type=float, default=36.0, help="newsroom: how far back to replay headlines") ap.add_argument("--gap", type=float, default=6.0, help="newsroom: seconds between replayed headlines") ap.add_argument("--dry", action="store_true", help="bot: Bybit prices but simulated fills, no Bybit key needed") ap.add_argument("--brain-every", type=float, default=10.0, help="bot: minutes between Claude's big-picture reads") ap.add_argument("--port", type=int, default=8765, help="dashboard port") ap.add_argument("--no-open", action="store_true", help="don't open the dashboard in a browser") a = ap.parse_args() from .core import console if a.command != "bot": # the bot prints its own banner, with its trading mode console.print("[bold #8b7bff]jev-starter[/] [dim]· paper trading on live Hyperliquid data · no real orders are ever placed[/]") if a.command == "check": check() return if a.command == "loop": from .loop import run_loop run_loop(a.coin.upper(), a.pace, a.minutes, a.port, not a.no_open, a.late_ms, not a.taker, a.maker_wait) elif a.command == "bot": from .bot import run_bot run_bot(a.coin.upper(), a.pace, a.minutes, a.port, not a.no_open, a.late_ms, a.maker_wait, a.brain_every, a.dry) if a.no_open and not a.minutes: return else: from .newsroom import run_newsroom run_newsroom(a.hours, a.gap, a.port, not a.no_open) console.print(" [dim]dashboard still open · Ctrl+C to stop[/]") try: threading.Event().wait() except KeyboardInterrupt: pass if __name__ == "__main__": main() ```` ### `jev-starter/jevlab/core.py` ````python """Shared bits: where results go, and the terminal styling.""" from __future__ import annotations from pathlib import Path from rich import box from rich.console import Console from rich.panel import Panel RESULTS = Path(__file__).resolve().parent.parent / "results" console = Console(highlight=False) def header(title: str, subtitle: str) -> None: console.print() console.print(Panel(f"[dim]{subtitle}[/]", title=f"[bold #8b7bff]{title}[/]", title_align="left", border_style="#3a3f5c", box=box.ROUNDED, padding=(0, 2))) ```` ### `jev-starter/jevlab/hl.py` ````python """Hyperliquid public market data. Mainnet, read-only, no key, no orders. Fills in every lab are simulated against this real data with Hyperliquid's real base-tier perp fees. Nothing here can place an order. """ from __future__ import annotations import time import pandas as pd import requests INFO_URL = "https://api.hyperliquid.xyz/info" # Hyperliquid perps, base fee tier (per side). TAKER_FEE = 0.00045 MAKER_FEE = 0.00015 def _info(body: dict, timeout: float = 10.0): r = requests.post(INFO_URL, json=body, timeout=timeout) r.raise_for_status() return r.json() def l2_book(coin: str) -> dict: """Best 20 levels a side, as floats: {"bids": [(px, sz)], "asks": [...], "time": ms}.""" d = _info({"type": "l2Book", "coin": coin}) bids, asks = d["levels"] return { "bids": [(float(l["px"]), float(l["sz"])) for l in bids], "asks": [(float(l["px"]), float(l["sz"])) for l in asks], "time": d["time"], } def mids(coins: list[str]) -> dict[str, float]: d = _info({"type": "allMids"}) return {c: float(d[c]) for c in coins if c in d} def candles(coin: str, interval: str, start_ms: int, end_ms: int | None = None) -> pd.DataFrame: """OHLCV candles, indexed by open time (UTC). Pages through the 5000-bar cap.""" end_ms = end_ms or int(time.time() * 1000) rows, cursor = [], start_ms while cursor < end_ms: batch = _info( {"type": "candleSnapshot", "req": {"coin": coin, "interval": interval, "startTime": cursor, "endTime": end_ms}} ) if not batch: break rows.extend(batch) last_open = batch[-1]["t"] if last_open <= cursor or len(batch) < 2: break cursor = last_open + 1 if not rows: return pd.DataFrame(columns=["open", "high", "low", "close", "volume"]) df = pd.DataFrame(rows).drop_duplicates("t") df.index = pd.to_datetime(df["t"], unit="ms", utc=True) out = df[["o", "h", "l", "c", "v"]].astype(float) out.columns = ["open", "high", "low", "close", "volume"] return out.sort_index() ```` ### `jev-starter/jevlab/judges.py` ````python """The Jev client. Sends one state plus typed questions, gets typed answers back. Jev question types (TypeSafe systemone API): noul yes/no, answered as a probability of "yes" choice pick one option from a dict of options score pick a level from a list ordered low -> high Answers are normalised to: noul -> {"p": float} choice -> {"choice": str, "probs": {option: float}} score -> {"score": float (0-based level, may be an expected value), "probs": [float, ...]} Reaches Jev through the Vercel AI Gateway (AI_GATEWAY_API_KEY), or directly with a TypeSafe key (TYPESAFE_API_KEY) if you have one. """ from __future__ import annotations import os import random import time import requests from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env")) JEV_GATEWAY_URL = "https://ai-gateway.vercel.sh/typesafe/v1/systemone" JEV_DIRECT_URL = "https://api.typesafe.ai/v1/systemone" class JudgeError(Exception): pass def _post(url: str, key: str, body: dict, timeout: float, retries: int = 6) -> dict: for attempt in range(retries + 1): retry_after = None try: r = requests.post( url, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, json=body, timeout=timeout, ) except requests.RequestException as exc: if attempt == retries: raise JudgeError(str(exc)) from exc else: if r.status_code == 200: return r.json() if r.status_code not in (429, 500, 502, 503, 529) or attempt == retries: raise JudgeError(f"HTTP {r.status_code}: {r.text[:300]}") retry_after = r.headers.get("retry-after") # 429 = the free tier is busy. Back off (honouring Retry-After) rather than drop the call. try: wait = float(retry_after) except (TypeError, ValueError): wait = min(8.0, 0.75 * 2**attempt) + random.random() time.sleep(wait) raise JudgeError("unreachable") # ---------------------------------------------------------------- normalising def _norm_probs(raw, options: list[str]) -> dict: if isinstance(raw, dict): probs = {o: float(raw.get(o, 0.0)) for o in options} elif isinstance(raw, list) and len(raw) == len(options): probs = dict(zip(options, map(float, raw))) else: return {} total = sum(probs.values()) return {o: p / total for o, p in probs.items()} if total > 0 else {} def normalise(questions: dict, raw_answers: dict) -> dict: """Coerce any judge's raw answers into the one shape the labs score.""" out = {} for qid, q in questions.items(): a = raw_answers.get(qid) if a is None: raise JudgeError(f"missing answer for '{qid}'") if q["type"] == "noul": p = a.get("noul", a.get("p")) if isinstance(a, dict) else a out[qid] = {"p": max(0.0, min(1.0, float(p)))} elif q["type"] == "choice": options = list(q["criteria"]) probs = _norm_probs(a.get("probabilities", a.get("probs")), options) choice = a.get("choice") if choice not in options: if not probs: raise JudgeError(f"bad choice answer for '{qid}': {a}") choice = max(probs, key=probs.get) if not probs: probs = {o: float(o == choice) for o in options} out[qid] = {"choice": choice, "probs": probs} elif q["type"] == "score": # Jev returns score as an expected level (e.g. 1.69) with probabilities # keyed "0".."n-1"; other judges may return a label or an index. levels = q["criteria"] score = a.get("score") if isinstance(score, str): score = levels.index(score) if score in levels else float(score) raw_p = a.get("probabilities", a.get("probs")) if isinstance(raw_p, dict) and all(str(k).isdigit() for k in raw_p): raw_p = {levels[int(k)]: v for k, v in raw_p.items() if int(k) < len(levels)} probs_d = _norm_probs(raw_p, levels) probs = [probs_d.get(l, 0.0) for l in levels] if probs_d else [ float(i == round(float(score))) for i in range(len(levels)) ] out[qid] = {"score": max(0.0, min(len(levels) - 1.0, float(score))), "probs": probs} return out # ---------------------------------------------------------------- judges class JevJudge: name = "jev" def __init__(self): direct = os.getenv("TYPESAFE_API_KEY", "").strip() gateway = os.getenv("AI_GATEWAY_API_KEY", "").strip() if direct: self.url, self.key, self.model = JEV_DIRECT_URL, direct, "jev-latest" elif gateway: self.url, self.key, self.model = JEV_GATEWAY_URL, gateway, "typesafe-ai/jev" else: raise JudgeError("no AI_GATEWAY_API_KEY or TYPESAFE_API_KEY in .env") def ask(self, state, questions, timeout=20.0, retries=6): t0 = time.monotonic() body = {"state": state, "model": self.model, "questions": questions} data = _post(self.url, self.key, body, timeout, retries) ms = (time.monotonic() - t0) * 1000 answers = normalise(questions, data.get("answers", data)) return answers, {"judge": self.name, "model": data.get("model", self.model), "latency_ms": round(ms)} ```` ### `jev-starter/jevlab/news.py` ````python """Crypto headlines from public RSS feeds, and the questions Jev answers about each one.""" from __future__ import annotations import html import re import xml.etree.ElementTree as ET from email.utils import parsedate_to_datetime import pandas as pd import requests FEEDS = { "cointelegraph": "https://cointelegraph.com/rss", "coindesk": "https://www.coindesk.com/arc/outboundfeeds/rss/", "decrypt": "https://decrypt.co/feed", "theblock": "https://www.theblock.co/rss.xml", } COINS = ["BTC", "ETH", "SOL"] QUESTIONS = { "asset": { "type": "choice", "instructions": "Which tradable asset does this news most directly affect?", "criteria": {"BTC": None, "ETH": None, "SOL": None, "whole_crypto_market": None, "none": None}, }, "direction": { "type": "choice", "instructions": "Is this news likely to push that asset's price up or down from here?", "criteria": {"bullish": None, "bearish": None, "neutral": None}, }, "impact": { "type": "score", "instructions": "How much is this news likely to move that asset's price?", "criteria": ["Noise", "Minor", "Notable", "Major"], }, } def fetch_headlines(hours: float, quiet: bool = False) -> list[dict]: cutoff = pd.Timestamp.now(tz="UTC") - pd.Timedelta(hours=hours) items, seen = [], set() for source, url in FEEDS.items(): try: r = requests.get(url, timeout=15, headers={"User-Agent": "jev-lab/0.1 rss reader"}) r.raise_for_status() root = ET.fromstring(r.content) except Exception as exc: if not quiet: print(f" ! feed {source} unavailable: {str(exc)[:80]}") continue n = 0 for it in root.iter("item"): title = html.unescape((it.findtext("title") or "").strip()) pub = it.findtext("pubDate") if not title or not pub or title.lower() in seen: continue ts = pd.Timestamp(parsedate_to_datetime(pub)).tz_convert("UTC") if ts < cutoff: continue summary = re.sub(r"<[^>]+>", " ", html.unescape(it.findtext("description") or "")) summary = re.sub(r"\s+", " ", summary).strip()[:400] seen.add(title.lower()) items.append({"source": source, "headline": title, "summary": summary, "published": ts}) n += 1 if not quiet: print(f" feed {source:<14} {n} headlines") return sorted(items, key=lambda x: x["published"]) def judge_call(ans: dict) -> tuple[str | None, int]: asset = ans["asset"]["choice"] coin = {"BTC": "BTC", "ETH": "ETH", "SOL": "SOL", "whole_crypto_market": "BTC"}.get(asset) side = {"bullish": 1, "bearish": -1}.get(ans["direction"]["choice"], 0) if ans["impact"]["score"] < 2: # only act on "Notable" or "Major" side = 0 return coin, side ```` ### `jev-starter/jevlab/server.py` ````python """A tiny local web server for the dashboards. Serves this folder on 127.0.0.1 only.""" from __future__ import annotations import threading import webbrowser from functools import partial from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent.parent class _Handler(SimpleHTTPRequestHandler): page = "loop.html" # which dashboard "/" opens def end_headers(self): self.send_header("Cache-Control", "no-store") super().end_headers() def do_GET(self): if self.path in ("/", "/index.html"): self.path = f"/dashboard/{self.page}" return super().do_GET() def log_message(self, *args): # keep the terminal clean pass def serve(port: int = 8765, open_browser: bool = True, page: str = "loop.html") -> ThreadingHTTPServer: handler = type("Handler", (_Handler,), {"page": page}) try: server = ThreadingHTTPServer(("127.0.0.1", port), partial(handler, directory=str(ROOT))) except OSError: raise SystemExit(f" port {port} is busy. Stop the other dashboard (Ctrl+C) or add --port {port + 1}") threading.Thread(target=server.serve_forever, daemon=True).start() url = f"http://127.0.0.1:{port}/" print(f" dashboard: {url}") if open_browser: webbrowser.open(url) return server ```` ### `jev-starter/jevlab/strategy.py` ````python """YOUR STRATEGY. This is the one file you're meant to edit. Jev makes a call several times a second ("buy, 91% sure"). Your strategy decides which of those calls actually become trades. Most should end in "hold": every trade has a cost, so a bot that trades every call bleeds. decide() is called every time Jev answers, with: call Jev's answer, e.g. {"side": "buy", "conf": 0.91} market the live numbers Jev was shown, for example: return_5s_bps, return_30s_bps price change over 5s / 30s (1 bps = 0.01%) top_of_book_imbalance -1 (all sellers) .. +1 (all buyers) at the best price depth5_imbalance the same, over the top 5 price levels aggressor_buy_share_5s / _30s share of recent trade volume that was buyers (0..1) trades_last_5s how busy the market is right now spread_bps, microprice_vs_mid_bps, tick_volatility_60s_bps claude_bias 24/7 bot only: "long", "short" or "flat", Claude's big-picture call, refreshed every few minutes position 1 = you're long, -1 = you're short, 0 = flat seconds_since_trade seconds since your last fill Return one of: "buy" go (or stay) long "sell" go (or stay) short "flat" close the position "hold · reason" do nothing. The reason shows up in the dashboard feed. The loop handles everything else: placing the (paper) order, fees, the dashboard. Want a different strategy? Describe it to Claude in one sentence and ask it to rewrite decide(), e.g. "only buy when Jev is 90%+ sure AND buyers have been in control for the last 30 seconds". """ # The defaults are the three rules from the video. SETTINGS = { "min_conf": 0.85, # only act when Jev is at least this sure "min_hold": 15, # seconds to sit still after a trade (no flip-flopping) } DESCRIPTION = f"trade only at ≥{SETTINGS['min_conf']:.0%} conviction · ≥{SETTINGS['min_hold']}s between flips" def decide(call: dict, market: dict, position: int, seconds_since_trade: float) -> str: want = 1 if call["side"] == "buy" else -1 bias = market.get("claude_bias") # only set on the 24/7 bot, where Claude is the brain if bias == "flat": return "flat" if position else "hold · Claude says stay out" if (bias == "long" and position < 0) or (bias == "short" and position > 0): return "flat" # Claude changed its mind: get out of the old direction first if (bias == "long" and want < 0) or (bias == "short" and want > 0): return f"hold · against Claude's {bias} bias" if call["conf"] < SETTINGS["min_conf"]: return "hold · low conviction" if want == position: return "hold · already " + ("long" if want > 0 else "short") if seconds_since_trade < SETTINGS["min_hold"]: return "hold · too soon to flip" return call["side"] # --------------------------------------------------------------------------- # Example: trade with the trend only. Uncomment to use it (and delete the # decide() above). # # def decide(call, market, position, seconds_since_trade): # trend_up = market.get("return_30s_bps", 0) > 0 and market.get("aggressor_buy_share_30s", 0.5) > 0.55 # trend_down = market.get("return_30s_bps", 0) < 0 and market.get("aggressor_buy_share_30s", 0.5) < 0.45 # if call["conf"] < 0.8: # return "hold · low conviction" # if call["side"] == "buy" and trend_up and position != 1: # return "buy" # if call["side"] == "sell" and trend_down and position != -1: # return "sell" # return "hold · against the trend" ```` ### `jev-starter/jevlab/loop.py` ````python """The Jev loop: Jev trading calls landing on a live Hyperliquid price feed, with YOUR strategy (jevlab/strategy.py) deciding which calls become trades. uv run python -m jevlab loop # HYPE, 10 minutes uv run python -m jevlab loop --coin BTC --minutes 0 # 0 = run until Ctrl+C The dashboard plots the live microprice (size-weighted mid) from Hyperliquid's best-bid/ask stream, about 8 updates a second. Jev is asked "buy or sell?" every --pace seconds (default 0.3s). Calls overlap, each answer lands the moment it arrives, and the pace backs off by itself if the key gets rate-limited (the free Vercel tier allows roughly one call every 2 seconds; $5 of credits removes that). Every Jev answer goes to strategy.decide(). If it says "buy" or "sell", the loop rests a limit order at the best bid/ask (maker fee, no spread). The order counts as filled only when the market trades THROUGH its price, and it's cancelled after --maker-wait seconds. --taker trades immediately at the ask/bid instead. Positions are $1,000, P&L is shown before and after fees, and nothing is hidden. PAPER ONLY: this reads public market data and simulates fills. It never places a real order and never needs an exchange key. """ from __future__ import annotations import json import os import statistics import threading import time from collections import deque from concurrent.futures import ThreadPoolExecutor import websocket from . import hl from .core import RESULTS, console, header from . import strategy from .judges import JevJudge, JudgeError from .server import serve WS_URL = "wss://api.hyperliquid.xyz/ws" NOTIONAL = 1000.0 # $ per side of the paper position TICK_KEEP_S = 300 # price history kept for the dashboard QUESTIONS = { "side": { "type": "choice", "instructions": "Which side should this bot hold for the next few seconds: buy (be long) or sell (be short)?", "criteria": {"buy": None, "sell": None}, } } class Market: """Live Hyperliquid feed for one coin: best bid/ask (~8 updates a second), top-5 depth, and every trade. Runs in a background thread and reconnects.""" def __init__(self, coin: str): self.coin = coin self.lock = threading.Lock() self.bbo = None # (bid, ask, bid_sz, ask_sz, t) self.depth5 = None # (bid_usd, ask_usd, t) self.trades = deque() # (t, is_buy, px, sz) self.mids = deque() # (t, mid) on every bbo update self.ticks = deque() # (t, microprice, bid, ask), thinned to 20/s for the dashboard self.ready = threading.Event() def start(self) -> None: def on_open(ws): for sub in ("bbo", "l2Book", "trades"): ws.send(json.dumps({"method": "subscribe", "subscription": {"type": sub, "coin": self.coin}})) self.app = websocket.WebSocketApp(WS_URL, on_open=on_open, on_message=lambda ws, m: self._on_message(m)) threading.Thread(target=lambda: self.app.run_forever(ping_interval=20, reconnect=3), daemon=True).start() def _on_message(self, raw: str) -> None: msg = json.loads(raw) ch, data, now = msg.get("channel"), msg.get("data"), time.time() with self.lock: if ch == "bbo" and data.get("bbo") and None not in data["bbo"]: b, a = data["bbo"] self.bbo = (float(b["px"]), float(a["px"]), float(b["sz"]), float(a["sz"]), now) bid, ask, bsz, asz, _ = self.bbo self.mids.append((now, (bid + ask) / 2)) if not self.ticks or now - self.ticks[-1][0] >= 0.05: self.ticks.append((now, (bid * asz + ask * bsz) / (bsz + asz), bid, ask)) self.ready.set() elif ch == "l2Book": bids, asks = data["levels"] self.depth5 = (sum(float(l["px"]) * float(l["sz"]) for l in bids[:5]), sum(float(l["px"]) * float(l["sz"]) for l in asks[:5]), now) elif ch == "trades": for tr in data: self.trades.append((tr["time"] / 1000, tr["side"] == "B", float(tr["px"]), float(tr["sz"]))) for dq, keep in ((self.mids, 120), (self.trades, 120), (self.ticks, TICK_KEEP_S)): while dq and now - dq[0][0] > keep: dq.popleft() def snapshot(self) -> dict: """The state Jev sees: small, numeric, and fresh. No dates, no coin name.""" with self.lock: bid, ask, bsz, asz, _ = self.bbo now = time.time() mid = (bid + ask) / 2 micro = (bid * asz + ask * bsz) / (bsz + asz) def ret(sec): past = next((m for t, m in self.mids if t >= now - sec), mid) return round(1e4 * (mid / past - 1), 2) def flow(sec): tr = [(b, px * sz) for t, b, px, sz in self.trades if t >= now - sec] vol = sum(v for _, v in tr) return (round(sum(v for b, v in tr if b) / vol, 3) if vol else 0.5), len(tr) buy5, n5 = flow(5) buy30, _ = flow(30) m60 = [m for t, m in self.mids if t >= now - 60] steps = [1e4 * (b / a - 1) for a, b in zip(m60, m60[1:]) if a != b] state = { "spread_bps": round(1e4 * (ask - bid) / mid, 3), "microprice_vs_mid_bps": round(1e4 * (micro / mid - 1), 3), "top_of_book_imbalance": round((bsz - asz) / (bsz + asz), 3), "aggressor_buy_share_5s": buy5, "aggressor_buy_share_30s": buy30, "trades_last_5s": n5, "return_5s_bps": ret(5), "return_30s_bps": ret(30), "tick_volatility_60s_bps": round(statistics.pstdev(steps), 3) if len(steps) > 2 else 0.0, } if self.depth5: bd, ad, _ = self.depth5 state["depth5_imbalance"] = round((bd - ad) / (bd + ad), 3) return {"mid": mid, "micro": micro, "bid": bid, "ask": ask, "spread_bps": 1e4 * (ask - bid) / mid, "state": state} def trades_since(self, t0: float) -> list[tuple]: with self.lock: return [tr for tr in self.trades if tr[0] >= t0] def recent_ticks(self, seconds: float = 90) -> list[list[float]]: with self.lock: now = time.time() return [[round(t, 3), m, b, a] for t, m, b, a in self.ticks if t >= now - seconds] class PaperBook: """A $1,000 long-or-short paper position. Tracks coins and cash, so buying at the bid or selling at the ask is credited exactly, and fees are kept apart.""" def __init__(self): self.pos, self.qty, self.cash, self.fees = 0, 0.0, 0.0, 0.0 self.trades = 0 self.equity = [] # [t, before_fees_usd, after_fees_usd] def fill(self, target: int, px: float, fee_rate: float) -> None: dq = target * NOTIONAL / px - self.qty self.cash -= dq * px self.fees += abs(dq) * px * fee_rate self.qty += dq self.pos = target self.trades += 1 def gross(self, mid: float) -> float: return self.cash + self.qty * mid def snap(self, t: float, mid: float) -> dict: g = self.gross(mid) self.equity.append([round(t, 2), round(g, 3), round(g - self.fees, 3)]) del self.equity[:-2400] return {"pos": self.pos, "gross": round(g, 3), "fees": round(self.fees, 3), "net": round(g - self.fees, 3), "trades": self.trades, "equity": self.equity} def run_loop(coin: str, pace_s: float, minutes: float, port: int, open_browser: bool, late_ms: float = 1500, maker: bool = True, maker_wait: float = 10.0) -> None: exec_txt = f"limit orders ({hl.MAKER_FEE * 1e4:.1f} bps)" if maker else f"market orders ({hl.TAKER_FEE * 1e4:.1f} bps + spread)" header("THE JEV LOOP", f"{coin}-PERP · Jev calls as fast as the key allows (floor {pace_s:g}s) · " f"strategy: {strategy.DESCRIPTION} · {exec_txt} · " f"{'until Ctrl+C' if not minutes else f'{minutes:g} min'} · $1,000 paper position") try: jev = JevJudge() except JudgeError as exc: raise SystemExit(f" Jev key missing: {exc}. Add AI_GATEWAY_API_KEY to .env first.") market = Market(coin) market.start() if not market.ready.wait(15): raise SystemExit(" no price feed from Hyperliquid after 15s, check the connection") RESULTS.mkdir(exist_ok=True) out = RESULTS / "loop.json" log = open(RESULTS / "loop_log.jsonl", "a") book = PaperBook() decisions: list[dict] = [] fills: list[dict] = [] latencies: list[float] = [] counts = {"ok": 0, "late": 0, "throttled": 0, "error": 0} lock = threading.Lock() st = {"interval": pace_s, "streak": 0, "prev": None, "order": None, "last_trade_t": 0.0, "hits": 0, "scored": 0} started = time.time() serve(port, open_browser, page="loop.html") def check_order() -> None: """Fill the resting limit order only if the market traded through its price.""" o = st["order"] if not o: return now = time.time() buying = o["buying"] through = any((buying and px < o["px"]) or (not buying and px > o["px"]) for _, _, px, _ in market.trades_since(o["t"])) snap = market.snapshot() through |= (buying and snap["ask"] < o["px"]) or (not buying and snap["bid"] > o["px"]) if through: book.fill(o["target"], o["px"], hl.MAKER_FEE) fills.append({"t": now, "side": "buy" if buying else "sell", "px": o["px"], "kind": "limit", "call": o["call"], "wait_s": round(now - o["t"], 1)}) st["order"], st["last_trade_t"] = None, now elif now > o["expires"]: st["order"] = None def write(status: str) -> None: with lock: check_order() mid = market.snapshot()["mid"] recent = [d["t"] for d in decisions if d["t"] >= time.time() - 10] payload = { "status": status, "coin": coin, "pace": pace_s, "interval": round(st["interval"], 3), "rate_per_s": round(len(recent) / 10, 2), "late_ms": late_ms, "strategy_note": strategy.DESCRIPTION, "execution": "maker" if maker else "taker", "maker_wait": maker_wait, "started": started, "updated": time.time(), "model": jev.model, "notional": NOTIONAL, "maker_fee_bps": hl.MAKER_FEE * 1e4, "taker_fee_bps": hl.TAKER_FEE * 1e4, "counts": dict(counts), "blocks": len(decisions), "last_ms": latencies[-1] if latencies else None, "avg_ms": round(statistics.mean(latencies[-200:])) if latencies else None, "hits": st["hits"], "scored": st["scored"], "order": st["order"], "decisions": decisions[-200:], "fills": fills[-100:], "book": book.snap(time.time(), mid), "ticks": market.recent_ticks(), } tmp = out.with_suffix(".tmp") tmp.write_text(json.dumps(payload)) os.replace(tmp, out) def decide(rec: dict, now: dict) -> str: """Hand Jev's call to strategy.decide(), then turn its answer into a (paper) order. Runs under the lock.""" try: choice = strategy.decide({"side": rec["side"], "conf": rec["conf"]}, rec["state"], book.pos, time.time() - st["last_trade_t"]) except Exception as exc: # a broken strategy shouldn't kill the loop: show it and hold return f"hold · strategy error: {str(exc)[:60]}" if choice not in ("buy", "sell", "flat"): return str(choice or "hold") want = {"buy": 1, "sell": -1, "flat": 0}[choice] if want == book.pos: if st["order"] and st["order"]["target"] != want: st["order"] = None return "hold · already " + {1: "long", -1: "short", 0: "flat"}[want] if st["order"] and st["order"]["target"] == want: return "hold · limit order working" if maker: buying = want > book.pos px = now["bid"] if buying else now["ask"] st["order"] = {"target": want, "buying": buying, "px": px, "t": time.time(), "expires": time.time() + maker_wait, "call": rec["block"]} return f"limit {'buy' if buying else 'sell'} @ {px:g}" buying = want > book.pos px = now["ask"] if buying else now["bid"] book.fill(want, px, hl.TAKER_FEE) fills.append({"t": time.time(), "side": "buy" if buying else "sell", "px": px, "kind": "market", "call": rec["block"], "wait_s": 0}) st["last_trade_t"] = time.time() return f"market {choice} @ {px:g}" def ask(seq: int) -> None: snap = market.snapshot() rec = {"block": seq, "t_ask": time.time(), "state": snap["state"]} try: ans, meta = jev.ask(snap["state"], QUESTIONS, timeout=5.0, retries=0) side = ans["side"]["choice"] rec.update(side=side, conf=round(ans["side"]["probs"][side], 3), ms=meta["latency_ms"], status="ok" if meta["latency_ms"] <= late_ms else "late") except JudgeError as exc: rec.update(side=None, conf=None, ms=None, status="throttled" if "429" in str(exc) else "error", error=str(exc)[:160]) now = market.snapshot() # the market as it is when the answer ARRIVES rec.update(t=time.time(), mid=now["mid"], micro=now["micro"]) with lock: counts[rec["status"]] += 1 if rec["ms"]: latencies.append(rec["ms"]) if rec["status"] == "throttled": # back off hard on a 429, creep back on good answers st["interval"], st["streak"] = min(4.0, st["interval"] * 1.6), 0 elif rec["status"] == "ok": st["streak"] += 1 if st["streak"] >= 3: st["interval"] = max(pace_s, st["interval"] * 0.9) prev = st["prev"] if prev and now["mid"] != prev["mid"]: # was the previous call's direction right, up to this one? st["hits"] += (now["mid"] > prev["mid"]) == (prev["side"] == "buy") st["scored"] += 1 if rec["status"] == "ok": st["prev"] = rec rec["action"] = decide(rec, now) else: rec["action"] = "hold · late" if rec["status"] == "late" else rec["status"] decisions.append(rec) log.write(json.dumps(rec) + "\n") colour = {"buy": "#3fd68a", "sell": "#ff5d6c"}.get(rec["side"], "#f5b53d") console.print(f" {rec['block']:>5} [{colour}]{(rec['side'] or '-').upper():<5}[/] " f"{(rec['conf'] or 0):.2f} {rec['ms'] or '-':>5} ms [dim]{rec['action']}[/]") console.print(" [dim] # side conf ms action[/]") stop = threading.Event() def writer(): # fills resting orders and keeps the dashboard's price history fresh while not stop.is_set(): write("running") stop.wait(0.25) threading.Thread(target=writer, daemon=True).start() seq = 0 try: with ThreadPoolExecutor(max_workers=8) as pool: while not minutes or time.time() - started < minutes * 60: seq += 1 pool.submit(ask, seq) time.sleep(st["interval"]) except KeyboardInterrupt: pass stop.set() write("done") log.close() mid = market.snapshot()["mid"] g = book.gross(mid) console.print(f"\n {len(decisions)} Jev calls · {book.trades} trades · before fees ${g:+.2f} · fees ${book.fees:.2f} · " f"after fees [bold]{'[#3fd68a]' if g - book.fees > 0 else '[#ff5d6c]'}${g - book.fees:+.2f}[/][/]") console.print(f" direction right {st['hits']}/{st['scored']} · late {counts['late']} · throttled {counts['throttled']}") ```` ### `jev-starter/jevlab/newsroom.py` ````python """The Jev newsroom: real crypto headlines arriving on a news terminal, Jev reading each one live, and the real price move that followed. uv run python -m jevlab newsroom # replay the last 36h, one headline every 6s uv run python -m jevlab newsroom --gap 4 --hours 24 Why a replay: crypto news only lands a few times an hour, which is too slow to watch. So the newsroom replays the last day or two of real headlines in fast motion. Every headline is sent to Jev live as it appears on screen (real answer, real response time), and because those headlines already happened, the chart can show what the price actually did over the next hour. The headlines are newer than Jev's training data, and Jev never sees the publish time. The trading rule: trade only headlines Jev rates "Notable" or "Major", in the direction it calls, from 1 minute after publishing, held for 60 minutes. No orders are ever placed. """ from __future__ import annotations import json import os import time import pandas as pd from . import hl from .core import RESULTS, console, header from .judges import JevJudge, JudgeError from .news import COINS, QUESTIONS, fetch_headlines, judge_call from .server import serve BEFORE_MIN, AFTER_MIN = 10, 60 # the reaction chart spans publish-10m .. publish+60m def price_paths(items: list[dict]) -> None: """Attach each coin's % move around every headline, from real 1m candles.""" start = int((items[0]["published"] - pd.Timedelta(minutes=BEFORE_MIN + 5)).timestamp() * 1000) candles = {c: hl.candles(c, "1m", start) for c in COINS} for it in items: entry_t = it["published"].floor("min") + pd.Timedelta(minutes=1) it["paths"] = {} for c, df in candles.items(): if entry_t not in df.index: continue base = df.loc[entry_t, "open"] window = df.loc[entry_t - pd.Timedelta(minutes=BEFORE_MIN): entry_t + pd.Timedelta(minutes=AFTER_MIN), "close"] it["paths"][c] = [[round((ts - entry_t).total_seconds() / 60, 1), round(float(1e4 * (px / base - 1)), 2)] for ts, px in window.items()] def run_newsroom(hours: float, every: float, port: int, open_browser: bool) -> None: header("THE JEV NEWSROOM", f"replaying the last {hours:g}h of real crypto headlines · one every {every:g}s · " f"Jev reads each one live · then the real {AFTER_MIN}-minute price move") try: jev = JevJudge() except JudgeError as exc: raise SystemExit(f" Jev key missing: {exc}. Add AI_GATEWAY_API_KEY to .env first.") def load_items() -> list[dict]: now = pd.Timestamp.now(tz="UTC") fresh = [i for i in fetch_headlines(hours) if i["published"] + pd.Timedelta(minutes=AFTER_MIN + 2) < now] if fresh: console.print(f" {len(fresh)} headlines · loading the price moves after each one…") price_paths(fresh) return fresh items = load_items() if not items: raise SystemExit(" no headlines old enough to score yet") RESULTS.mkdir(exist_ok=True) out = RESULTS / "newsroom.json" score = {"read": 0, "tagged": 0, "trades": 0, "right": 0, "net_bps": 0.0, "latency_ms": []} feed: list[dict] = [] started = time.time() rnd = {"n": 1, "index": 0} def write(status: str) -> None: lat = sorted(score["latency_ms"]) payload = {"status": status, "hours": hours, "every": every, "total": len(items), "index": rnd["index"], "round": rnd["n"], "started": started, "updated": time.time(), "model": jev.model, "cost_bps": round(2e4 * hl.TAKER_FEE + 1, 1), "after_min": AFTER_MIN, "before_min": BEFORE_MIN, "score": {**{k: v for k, v in score.items() if k != "latency_ms"}, "median_ms": lat[len(lat) // 2] if lat else None}, "feed": feed[-300:]} tmp = out.with_suffix(".tmp") tmp.write_text(json.dumps(payload, default=str)) os.replace(tmp, out) write("running") serve(port, open_browser, page="newsroom.html") time.sleep(2.5) # let the page load before the first headline lands try: while True: for it in items: rnd["index"] += 1 t0 = time.time() row = {"headline": it["headline"], "source": it["source"], "published": it["published"].timestamp(), "paths": it["paths"], "arrived": t0, "answer": None} feed.append(row) write("running") # the headline hits the wire before Jev has answered state = {"headline": it["headline"], "summary": it["summary"], "source": it["source"]} try: ans, meta = jev.ask(state, QUESTIONS, timeout=10.0, retries=3) except JudgeError as exc: row.update(error=str(exc)[:120], answered=time.time()) write("running") console.print(f" [#f5b53d]skip[/] {it['headline'][:70]} ({str(exc)[:40]})") time.sleep(max(0.0, every - (time.time() - t0))) continue coin, side = judge_call(ans) side = side if coin else 0 # no tradeable coin, no trade path = it["paths"].get(coin or "BTC") or [] ret = next((v for m, v in path if m == AFTER_MIN), None) row.update(answer=ans, ms=meta["latency_ms"], answered=time.time(), coin=coin, direction=ans["direction"]["choice"], impact=ans["impact"]["score"], trade=bool(side), side=side, ret_bps=ret) score["read"] += 1 score["latency_ms"].append(meta["latency_ms"]) if coin: score["tagged"] += 1 if side and ret is not None: score["trades"] += 1 score["right"] += int(side * ret > 0) row["net_bps"] = round(side * ret - (2e4 * hl.TAKER_FEE + 1), 1) score["net_bps"] += row["net_bps"] write("running") tag = f"{coin or '—':<4} {row['direction']:<8} impact {row['impact']:.1f}" console.print(f" {meta['latency_ms']:>4} ms {tag} {'[bold]TRADE[/]' if side else '[dim]no trade[/]'} " f"[dim]{it['headline'][:60]}[/]") time.sleep(max(0.0, every - (time.time() - t0))) # round finished: pull the latest headlines and go again, with a fresh scoreboard console.print(f"\n round {rnd['n']} done · reloading headlines for the next round") items = load_items() or items rnd["n"] += 1 rnd["index"] = 0 score.update(read=0, tagged=0, trades=0, right=0, net_bps=0.0, latency_ms=[]) except KeyboardInterrupt: pass write("done") console.print(f"\n {score['read']} headlines read · {score['tagged']} tagged to a coin · {score['trades']} worth trading · " f"{score['right']} right · {score['net_bps']:+.1f} bps after costs") ```` ### `jev-starter/jevlab/brain.py` ````python """Claude as the big-picture brain (Path 2). Every few minutes, Claude reads a plain summary of the market (recent price moves, volatility, funding, and the latest crypto headlines) and sets the bias for the next stretch: long, short, or flat. Jev and strategy.py still make the fast calls, but only in the direction Claude allows. Uses the same Vercel AI Gateway key as Jev. The model is CLAUDE_MODEL in .env (default anthropic/claude-sonnet-5). Claude on the gateway needs AI Gateway credits (the paid tier). """ from __future__ import annotations import json import os import re import threading import time import requests from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env")) CHAT_URL = "https://ai-gateway.vercel.sh/v1/chat/completions" BYBIT_REST = "https://api.bybit.com/v5/market" def market_summary(symbol: str) -> dict: """Plain numbers Claude can reason about, from Bybit's public market data.""" k = requests.get(f"{BYBIT_REST}/kline", params={"category": "linear", "symbol": symbol, "interval": "5", "limit": 49}, timeout=10).json()["result"]["list"] closes = [float(r[4]) for r in reversed(k)] # oldest -> newest, 5-minute bars, last ~4 hours t = requests.get(f"{BYBIT_REST}/tickers", params={"category": "linear", "symbol": symbol}, timeout=10).json()["result"]["list"][0] pct = lambda a, b: round(100 * (b / a - 1), 2) moves = [abs(pct(a, b)) for a, b in zip(closes, closes[1:])] return { "symbol": symbol, "price": closes[-1], "change_15m_pct": pct(closes[-4], closes[-1]), "change_1h_pct": pct(closes[-13], closes[-1]), "change_4h_pct": pct(closes[0], closes[-1]), "change_24h_pct": round(100 * float(t["price24hPcnt"]), 2), "avg_5m_move_pct": round(sum(moves) / len(moves), 3), "high_4h": max(closes), "low_4h": min(closes), "funding_rate_pct": round(100 * float(t.get("fundingRate") or 0), 4), } def headlines(limit: int = 8) -> list[str]: try: from .news import fetch_headlines return [h["headline"] for h in fetch_headlines(6, quiet=True)][-limit:] except Exception: return [] PROMPT = """You are the portfolio manager for a small, cautious crypto trading bot. A fast AI makes buy/sell calls every second, but it can only trade in the direction you allow. Decide the bias for the next {minutes} minutes: "long" (only buying allowed), "short" (only selling allowed), or "flat" (stay out, close any position). If the evidence leans one way (momentum across timeframes, funding, the news), pick that side and let "confidence" say how strongly it leans. Choose "flat" only when the signals genuinely conflict or the market is dead. Market data: {summary} Latest crypto headlines: {news} Reply with ONLY a JSON object: {{"bias": "long" | "short" | "flat", "confidence": 0.0-1.0, "reason": "one short sentence"}}""" class Brain: def __init__(self, symbol: str, every_min: float): self.key = os.getenv("AI_GATEWAY_API_KEY", "").strip() self.model = os.getenv("CLAUDE_MODEL", "anthropic/claude-sonnet-5").strip() self.symbol, self.every = symbol, every_min self.state = {"bias": None, "confidence": None, "reason": "waiting for Claude's first read", "t": None, "model": self.model, "ms": None, "error": None} self.lock = threading.Lock() def think(self) -> None: t0 = time.time() try: summary = market_summary(self.symbol) news = "\n".join(f"- {h}" for h in headlines()) or "- (none)" body = {"model": self.model, "temperature": 0, "max_tokens": 200, "messages": [ {"role": "user", "content": PROMPT.format(minutes=self.every, summary=json.dumps(summary, indent=1), news=news)}]} r = requests.post(CHAT_URL, headers={"Authorization": f"Bearer {self.key}"}, json=body, timeout=60) if r.status_code != 200: raise RuntimeError(f"HTTP {r.status_code}: {r.text[:160]}") text = r.json()["choices"][0]["message"]["content"] out = json.loads(re.search(r"\{.*\}", text, re.S).group(0)) bias = out.get("bias") if out.get("bias") in ("long", "short", "flat") else "flat" with self.lock: self.state.update(bias=bias, confidence=float(out.get("confidence") or 0), reason=str(out.get("reason", ""))[:160], t=time.time(), ms=round((time.time() - t0) * 1000), error=None, summary=summary) except Exception as exc: # keep the last bias; if there's never been one, the bot stays flat with self.lock: self.state["error"] = str(exc)[:160] if self.state["bias"] is None: self.state.update(bias="flat", reason="Claude unavailable, staying out", t=time.time()) def start(self, stop: threading.Event) -> None: def run(): while not stop.is_set(): self.think() stop.wait(self.every * 60) threading.Thread(target=run, daemon=True).start() def current(self) -> dict: with self.lock: return dict(self.state) ```` ### `jev-starter/jevlab/bybit.py` ````python """Bybit: the live price feed and order placement for the 24/7 bot (Path 2). Two modes, set with BYBIT_MODE in .env: demo (default) Bybit Demo Trading: real market prices, fake money. live Real money. Refused unless JEV_LIVE_CONFIRM is set to the exact phrase in LIVE_PHRASE. Only the account owner should ever set it. Orders are post-only limit orders at the best bid (to buy) or best ask (to sell), so the bot never crosses the spread and pays the maker fee. An order that isn't filled within the wait time is cancelled. """ from __future__ import annotations import json import math import os import threading import time import websocket from dotenv import load_dotenv from pybit.unified_trading import HTTP from .loop import Market load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env")) WS_PUBLIC = "wss://stream.bybit.com/v5/public/linear" LIVE_PHRASE = "I accept the risk of trading real money" class BybitError(Exception): pass def bybit_mode() -> str: mode = os.getenv("BYBIT_MODE", "demo").strip().lower() or "demo" if mode not in ("demo", "live"): raise BybitError(f"BYBIT_MODE must be 'demo' or 'live', not '{mode}'") if mode == "live" and os.getenv("JEV_LIVE_CONFIRM", "").strip() != LIVE_PHRASE: raise BybitError("BYBIT_MODE=live, but JEV_LIVE_CONFIRM isn't set to the exact confirmation phrase. " "Refusing to trade real money.") return mode class BybitMarket(Market): """Same live snapshot as the Hyperliquid feed, from Bybit's public stream: best bid/ask (level 1 order book, ~25 updates a second) and every trade.""" def __init__(self, symbol: str): super().__init__(symbol) self.symbol = symbol def start(self) -> None: topics = [f"orderbook.1.{self.symbol}", f"publicTrade.{self.symbol}"] def on_open(ws): ws.send(json.dumps({"op": "subscribe", "args": topics})) def heartbeat(): # Bybit drops connections that don't ping every ~20s while ws.sock and ws.sock.connected: time.sleep(18) try: ws.send(json.dumps({"op": "ping"})) except Exception: return threading.Thread(target=heartbeat, daemon=True).start() self.app = websocket.WebSocketApp(WS_PUBLIC, on_open=on_open, on_message=lambda ws, m: self._on_bybit(m)) threading.Thread(target=lambda: self.app.run_forever(reconnect=3), daemon=True).start() def _on_bybit(self, raw: str) -> None: msg = json.loads(raw) topic, data, now = msg.get("topic", ""), msg.get("data"), time.time() with self.lock: if topic.startswith("orderbook.1.") and data and data.get("b") and data.get("a"): bid, bsz = map(float, data["b"][0]) ask, asz = map(float, data["a"][0]) self.bbo = (bid, ask, bsz, asz, now) self.mids.append((now, (bid + ask) / 2)) if not self.ticks or now - self.ticks[-1][0] >= 0.05: self.ticks.append((now, (bid * asz + ask * bsz) / (bsz + asz), bid, ask)) self.ready.set() elif topic.startswith("publicTrade.") and data: for tr in data: self.trades.append((tr["T"] / 1000, tr["S"] == "Buy", float(tr["p"]), float(tr["v"]))) for dq, keep in ((self.mids, 120), (self.trades, 120), (self.ticks, 300)): while dq and now - dq[0][0] > keep: dq.popleft() class BybitTrader: """Places and tracks post-only limit orders on one USDT perpetual.""" def __init__(self, symbol: str, mode: str): key, secret = os.getenv("BYBIT_API_KEY", "").strip(), os.getenv("BYBIT_API_SECRET", "").strip() if not key or not secret: raise BybitError("BYBIT_API_KEY / BYBIT_API_SECRET are missing from .env") self.symbol, self.mode = symbol, mode self.http = HTTP(demo=(mode == "demo"), api_key=key, api_secret=secret, recv_window=10000) info = HTTP().get_instruments_info(category="linear", symbol=symbol)["result"]["list"] if not info: raise BybitError(f"{symbol} isn't a Bybit USDT perpetual") lot, price = info[0]["lotSizeFilter"], info[0]["priceFilter"] self.qty_step, self.min_qty = float(lot["qtyStep"]), float(lot["minOrderQty"]) self.min_notional = float(lot.get("minNotionalValue") or 5) self.tick = float(price["tickSize"]) # --- helpers ----------------------------------------------------------------- def _decimals(self, step: float) -> int: return max(0, -int(math.floor(math.log10(step)))) if step < 1 else 0 def round_qty(self, qty: float) -> float: return round(math.floor(qty / self.qty_step + 1e-9) * self.qty_step, self._decimals(self.qty_step)) def fmt_px(self, px: float) -> str: return f"{round(round(px / self.tick) * self.tick, self._decimals(self.tick)):.{self._decimals(self.tick)}f}" def _call(self, fn, **kw) -> dict: try: res = fn(**kw) except Exception as exc: # pybit raises on non-zero retCode raise BybitError(str(exc)[:300]) from exc if res.get("retCode") not in (0, None): raise BybitError(f"{res.get('retCode')}: {res.get('retMsg')}") return res.get("result", {}) # --- account ------------------------------------------------------------------- def equity_usdt(self) -> float | None: res = self._call(self.http.get_wallet_balance, accountType="UNIFIED") for acct in res.get("list", []): if acct.get("totalEquity"): return float(acct["totalEquity"]) return None def position(self) -> tuple[float, float]: """Signed size in coins (+ long, - short) and average entry price.""" rows = self._call(self.http.get_positions, category="linear", symbol=self.symbol).get("list", []) size = sum(float(r["size"]) * (1 if r["side"] == "Buy" else -1) for r in rows if float(r.get("size") or 0)) avg = next((float(r["avgPrice"]) for r in rows if float(r.get("size") or 0)), 0.0) return size, avg # --- orders -------------------------------------------------------------------- def place_post_only(self, side: str, qty: float, px: float, reduce_only: bool = False) -> str: res = self._call(self.http.place_order, category="linear", symbol=self.symbol, side="Buy" if side == "buy" else "Sell", orderType="Limit", qty=str(qty), price=self.fmt_px(px), timeInForce="PostOnly", reduceOnly=reduce_only, orderLinkId=f"jev-{int(time.time() * 1000)}") return res["orderId"] def order(self, order_id: str) -> dict: """Current state of one order: status, filled qty, average price, fee paid.""" for fn in (self.http.get_open_orders, self.http.get_order_history): rows = self._call(fn, category="linear", symbol=self.symbol, orderId=order_id).get("list", []) if rows: r = rows[0] return {"status": r["orderStatus"], "filled": float(r.get("cumExecQty") or 0), "avg_px": float(r.get("avgPrice") or 0), "fee": float(r.get("cumExecFee") or 0)} return {"status": "Unknown", "filled": 0.0, "avg_px": 0.0, "fee": 0.0} def cancel(self, order_id: str) -> None: try: self._call(self.http.cancel_order, category="linear", symbol=self.symbol, orderId=order_id) except BybitError: pass # already filled or gone def cancel_all(self) -> None: try: self._call(self.http.cancel_all_orders, category="linear", symbol=self.symbol) except BybitError: pass def flatten(self) -> dict | None: """Close any open position with a reduce-only market order (kill switch, loss limit, exit). Returns the closing fill (signed qty, price, fee) so the P&L includes it.""" size, _ = self.position() if abs(size) < self.min_qty: return None res = self._call(self.http.place_order, category="linear", symbol=self.symbol, side="Sell" if size > 0 else "Buy", orderType="Market", qty=str(self.round_qty(abs(size))), reduceOnly=True) for _ in range(10): # market orders fill almost instantly; give it a moment to report time.sleep(0.3) o = self.order(res["orderId"]) if o["status"] == "Filled": return {"dq": -size, "px": o["avg_px"], "fee": o["fee"]} return {"dq": -size, "px": 0.0, "fee": 0.0} ```` ### `jev-starter/jevlab/bot.py` ````python """The 24/7 bot (Path 2): Jev + Claude + your strategy, trading on Bybit. uv run python -m jevlab bot --dry # Bybit prices, simulated fills. No Bybit key needed uv run python -m jevlab bot # Bybit Demo Trading (fake money), needs a demo API key uv run python -m jevlab bot --minutes 0 # run until stopped (this is what the server runs) How it decides: * Claude (the brain) reads the market every --brain-every minutes and sets the bias: long, short or flat. * Jev makes a fast buy/sell call several times a second. * strategy.py combines them. By default it only trades in Claude's direction, and only when Jev is 85%+ sure. * Orders are post-only limit orders at the best bid/ask (maker fee, no spread), cancelled if they don't fill within --maker-wait seconds. Safety, always on: * Demo Trading unless BYBIT_MODE=live AND JEV_LIVE_CONFIRM is the exact phrase (see bybit.py). * MAX_POSITION_USD caps the position size (default $1,000). * MAX_DAILY_LOSS_USD: hit it and the bot cancels everything, closes the position and stops trading for the day (default $50). * Kill switch: create a file called STOP in the project folder, and the bot closes out and halts. * On shutdown it cancels open orders and closes the position. """ from __future__ import annotations import json import os import statistics import threading import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from pathlib import Path from . import strategy from .brain import Brain from .bybit import BybitError, BybitMarket, BybitTrader, bybit_mode from .core import RESULTS, console, header from .judges import JevJudge, JudgeError from .loop import QUESTIONS from .server import serve MAKER_FEE = 0.0002 # Bybit base-tier maker fee on USDT perps, used for --dry fills STOP_FILE = Path(__file__).resolve().parent.parent / "STOP" class Book: """Position and P&L from actual fills: coins, cash and fees, marked to the live mid.""" def __init__(self): self.qty, self.cash, self.fees, self.trades = 0.0, 0.0, 0.0, 0 self.equity: list[list[float]] = [] self.day = datetime.now(timezone.utc).date() self.day_start_net = 0.0 def fill(self, dq: float, px: float, fee_usd: float) -> None: self.cash -= dq * px self.qty += dq self.fees += fee_usd self.trades += 1 def net(self, mid: float) -> float: return self.cash + self.qty * mid - self.fees def snap(self, t: float, mid: float) -> dict: g = self.cash + self.qty * mid self.equity.append([round(t, 2), round(g, 3), round(g - self.fees, 3)]) del self.equity[:-2400] pos = 1 if self.qty > 0 else -1 if self.qty < 0 else 0 return {"pos": pos, "qty": self.qty, "gross": round(g, 3), "fees": round(self.fees, 3), "net": round(g - self.fees, 3), "trades": self.trades, "equity": self.equity} def run_bot(coin: str, pace_s: float, minutes: float, port: int, open_browser: bool, late_ms: float = 1500, maker_wait: float = 10.0, brain_every: float = 10.0, dry: bool = False) -> None: symbol = f"{coin}USDT" try: mode = "dry" if dry else bybit_mode() except BybitError as exc: raise SystemExit(f" {exc}") position_usd = float(os.getenv("MAX_POSITION_USD", "1000")) max_loss = float(os.getenv("MAX_DAILY_LOSS_USD", "50")) mode_label = {"dry": "Bybit prices · simulated fills", "demo": "Bybit DEMO · fake money", "live": "LIVE · REAL MONEY"}[mode] header("THE JEV BOT", f"{symbol} on Bybit · {mode_label} · Claude sets the bias every {brain_every:g} min · " f"Jev calls as fast as the key allows · strategy: {strategy.DESCRIPTION} · " f"max position ${position_usd:,.0f} · daily loss limit ${max_loss:,.0f}") if mode == "live": console.print(" [bold #ff5d6c]LIVE MODE: this bot is trading real money.[/] Kill switch: create a file named STOP.") try: jev = JevJudge() except JudgeError as exc: raise SystemExit(f" Jev key missing: {exc}. Add AI_GATEWAY_API_KEY to .env first.") trader = None if mode != "dry": try: trader = BybitTrader(symbol, mode) trader.cancel_all() # start clean: no stale orders from a previous run equity = trader.equity_usdt() console.print(f" Bybit {mode} account connected · equity {equity:,.2f} USDT" if equity is not None else f" Bybit {mode} account connected") except BybitError as exc: raise SystemExit(f" Bybit problem: {exc}") market = BybitMarket(symbol) market.start() if not market.ready.wait(15): raise SystemExit(" no price feed from Bybit after 15s, check the connection") book = Book() if trader: # adopt any position already open, so the numbers are honest from the first second size, avg = trader.position() if size: book.qty, book.cash = size, -size * avg console.print(f" existing position adopted: {size:+g} {coin} @ {avg:g}") stop = threading.Event() brain = Brain(symbol, brain_every) brain.start(stop) RESULTS.mkdir(exist_ok=True) out = RESULTS / "loop.json" log = open(RESULTS / "bot_log.jsonl", "a") decisions: list[dict] = [] fills: list[dict] = [] latencies: list[float] = [] counts = {"ok": 0, "late": 0, "throttled": 0, "error": 0} lock = threading.Lock() st = {"interval": pace_s, "streak": 0, "prev": None, "order": None, "last_trade_t": 0.0, "hits": 0, "scored": 0, "halted": None} started = time.time() serve(port, open_browser, page="loop.html") def record_fill(dq: float, px: float, fee: float, kind: str, call, waited: float) -> None: book.fill(dq, px, fee) fills.append({"t": time.time(), "side": "buy" if dq > 0 else "sell", "px": px, "qty": abs(dq), "kind": kind, "call": call, "wait_s": round(waited, 1)}) st["last_trade_t"] = time.time() def check_order() -> None: o = st["order"] if not o: return now = time.time() if trader is None: # --dry: fill only if the market traded through our price through = any((o["buying"] and px < o["px"]) or (not o["buying"] and px > o["px"]) for _, _, px, _ in market.trades_since(o["t"])) if through: record_fill(o["qty"] if o["buying"] else -o["qty"], o["px"], o["qty"] * o["px"] * MAKER_FEE, "limit", o["call"], now - o["t"]) st["order"] = None elif now > o["expires"]: st["order"] = None return try: s = trader.order(o["id"]) except BybitError: return new = s["filled"] - o["seen_qty"] if new > 1e-12: # count only the newly filled part, with its share of the fee px = s["avg_px"] or o["px"] record_fill(new if o["buying"] else -new, px, s["fee"] - o["seen_fee"], "limit", o["call"], now - o["t"]) o["seen_qty"], o["seen_fee"] = s["filled"], s["fee"] if s["status"] in ("Filled", "Cancelled", "Rejected", "Deactivated", "PartiallyFilledCanceled"): st["order"] = None elif now > o["expires"]: trader.cancel(o["id"]) o["expires"] = now + 3 # look once more for fills that raced the cancel, then let go def close_out() -> None: try: f = trader.flatten() except BybitError as exc: console.print(f" [#ff5d6c]couldn't close the position: {exc}. Close it on Bybit manually.[/]") return if f and f["px"]: record_fill(f["dq"], f["px"], f["fee"], "market", None, 0) def halt(reason: str) -> None: st["halted"] = reason st["order"] = None console.print(f"\n [bold #f5b53d]HALTED: {reason}[/]. Cancelling orders and closing the position.") if trader: trader.cancel_all() close_out() else: mid = market.snapshot()["mid"] if book.qty: record_fill(-book.qty, mid, abs(book.qty) * mid * 0.00055, "market", None, 0) def risk_check(mid: float) -> None: if st["halted"]: return today = datetime.now(timezone.utc).date() if today != book.day: # a new UTC day resets the loss limit book.day, book.day_start_net = today, book.net(mid) if book.net(mid) - book.day_start_net < -max_loss: halt(f"daily loss limit (${max_loss:,.0f}) reached") elif STOP_FILE.exists(): halt("kill switch (STOP file)") def write(status: str) -> None: with lock: check_order() mid = market.snapshot()["mid"] risk_check(mid) recent = [d["t"] for d in decisions if d["t"] >= time.time() - 10] o = st["order"] payload = { "status": status, "venue": "bybit", "symbol": symbol, "coin": coin, "mode": mode, "venue_label": f"{symbol} · Bybit", "mode_label": mode_label + (f" · HALTED: {st['halted']}" if st["halted"] else ""), "pace": pace_s, "interval": round(st["interval"], 3), "rate_per_s": round(len(recent) / 10, 2), "late_ms": late_ms, "strategy_note": strategy.DESCRIPTION, "execution": "maker", "maker_wait": maker_wait, "started": started, "updated": time.time(), "model": jev.model, "notional": position_usd, "maker_fee_bps": MAKER_FEE * 1e4, "taker_fee_bps": 5.5, "counts": dict(counts), "blocks": len(decisions), "last_ms": latencies[-1] if latencies else None, "avg_ms": round(statistics.mean(latencies[-200:])) if latencies else None, "hits": st["hits"], "scored": st["scored"], "brain": brain.current(), "order": ({"target": o["target"], "buying": o["buying"], "px": o["px"], "t": o["t"]} if o else None), "decisions": decisions[-200:], "fills": fills[-100:], "book": book.snap(time.time(), mid), "ticks": market.recent_ticks(), } tmp = out.with_suffix(".tmp") tmp.write_text(json.dumps(payload, default=str)) os.replace(tmp, out) def decide(rec: dict, now: dict) -> str: """Jev's call + Claude's bias -> strategy.decide() -> a post-only limit order. Runs under the lock.""" if st["halted"]: return f"hold · halted ({st['halted']})" bias = brain.current().get("bias") if bias is None: return "hold · waiting for Claude's first read" market_view = {**rec["state"], "claude_bias": bias} pos = 1 if book.qty > 0 else -1 if book.qty < 0 else 0 try: choice = strategy.decide({"side": rec["side"], "conf": rec["conf"]}, market_view, pos, time.time() - st["last_trade_t"]) except Exception as exc: return f"hold · strategy error: {str(exc)[:60]}" if choice not in ("buy", "sell", "flat"): return str(choice or "hold") want = {"buy": 1, "sell": -1, "flat": 0}[choice] if want == pos: return "hold · already " + {1: "long", -1: "short", 0: "flat"}[want] if st["order"]: if st["order"]["target"] == want: return "hold · limit order working" if trader: trader.cancel(st["order"]["id"]) st["order"] = None size = position_usd / now["mid"] size = trader.round_qty(size) if trader else round(size, 6) dq = want * size - book.qty buying = dq > 0 qty = trader.round_qty(abs(dq)) if trader else round(abs(dq), 6) if trader and (qty < trader.min_qty or qty * now["mid"] < trader.min_notional): return "hold · order too small for Bybit" px = now["bid"] if buying else now["ask"] order = {"target": want, "buying": buying, "qty": qty, "px": px, "t": time.time(), "expires": time.time() + maker_wait, "call": rec["block"], "seen_qty": 0.0, "seen_fee": 0.0, "id": None} if trader: try: order["id"] = trader.place_post_only("buy" if buying else "sell", qty, px, reduce_only=(want == 0)) except BybitError as exc: return f"hold · order rejected: {str(exc)[:70]}" st["order"] = order return f"limit {'buy' if buying else 'sell'} {qty:g} @ {px:g}" def ask(seq: int) -> None: snap = market.snapshot() rec = {"block": seq, "t_ask": time.time(), "state": snap["state"]} try: ans, meta = jev.ask(snap["state"], QUESTIONS, timeout=5.0, retries=0) side = ans["side"]["choice"] rec.update(side=side, conf=round(ans["side"]["probs"][side], 3), ms=meta["latency_ms"], status="ok" if meta["latency_ms"] <= late_ms else "late") except JudgeError as exc: rec.update(side=None, conf=None, ms=None, status="throttled" if "429" in str(exc) else "error", error=str(exc)[:160]) now = market.snapshot() rec.update(t=time.time(), mid=now["mid"], micro=now["micro"]) with lock: counts[rec["status"]] += 1 if rec["ms"]: latencies.append(rec["ms"]) if rec["status"] == "throttled": st["interval"], st["streak"] = min(4.0, st["interval"] * 1.6), 0 elif rec["status"] == "ok": st["streak"] += 1 if st["streak"] >= 3: st["interval"] = max(pace_s, st["interval"] * 0.9) prev = st["prev"] if prev and now["mid"] != prev["mid"]: st["hits"] += (now["mid"] > prev["mid"]) == (prev["side"] == "buy") st["scored"] += 1 if rec["status"] == "ok": st["prev"] = rec rec["action"] = decide(rec, now) else: rec["action"] = "hold · late" if rec["status"] == "late" else rec["status"] decisions.append(rec) log.write(json.dumps(rec) + "\n") log.flush() if rec["action"].startswith("limit") or rec["action"].startswith("hold · order"): console.print(f" {time.strftime('%H:%M:%S')} #{rec['block']:<6} Jev {(rec['side'] or '-').upper():<4} " f"{(rec['conf'] or 0):.2f} → {rec['action']}") def writer(): while not stop.is_set(): try: write("running") except Exception as exc: # a hiccup reading the exchange shouldn't stop the bot console.print(f" [dim]status update skipped: {str(exc)[:80]}[/]") stop.wait(0.5) threading.Thread(target=writer, daemon=True).start() console.print(" [dim]trades and orders print here · the dashboard shows every call[/]") seq = 0 try: with ThreadPoolExecutor(max_workers=8) as pool: while not minutes or time.time() - started < minutes * 60: seq += 1 pool.submit(ask, seq) time.sleep(st["interval"]) except KeyboardInterrupt: pass stop.set() with lock: if trader: trader.cancel_all() if os.getenv("FLATTEN_ON_EXIT", "true").lower() != "false": close_out() write("done") log.close() mid = market.snapshot()["mid"] console.print(f"\n {len(decisions)} Jev calls · {book.trades} fills · fees ${book.fees:.2f} · net ${book.net(mid):+.2f}") ```` ### `jev-starter/deploy/jev-bot.service` ````ini # systemd service: keeps the Jev bot running 24/7 on the server and restarts it if it crashes. # Install: sudo cp deploy/jev-bot.service /etc/systemd/system/ && sudo systemctl daemon-reload # sudo systemctl enable --now jev-bot # Logs: journalctl -u jev-bot -f Stop: sudo systemctl stop jev-bot [Unit] Description=Jev trading bot (Bybit) After=network-online.target Wants=network-online.target [Service] WorkingDirectory=/root/jev-starter ExecStart=/root/.local/bin/uv run python -m jevlab bot --minutes 0 --no-open Restart=on-failure RestartSec=15 KillSignal=SIGINT TimeoutStopSec=30 [Install] WantedBy=multi-user.target ```` ### `jev-starter/dashboard/loop.html` ````html Jev Loop
Jev Loop
paper mode
—
typesafe-ai/jev
00:00:00
—
Waiting
first block…
what Jev saw
connecting to Hyperliquid…
one dot per decision · buy · sell · late · ◆ = trade filled
Jev's latest call
—
buy
– sell
–
Paper account · $1,000 position
Paper balance$1,000.00
Before fees– Fees paid –
After fees– Jev calls → trades– Direction right–
Feed · every Jev call
#callconfmsaction
```` ### `jev-starter/dashboard/newsroom.html` ````html Jev Newsroom
Jev Newsroom
REPLAY · real headlines
0 / 0
typesafe-ai/jev
Newswirecointelegraph · coindesk · decrypt · the block
Jev's read
reading…

Which coin?

Which way?

How big?

score–
waiting for the first headline…
Headlines read0
Tagged to a coin0
Worth trading (Notable+)0
Direction right–
After costs, 60-min holds–
````