""" ZEC Renko Line Bot — PAPER TRADING ONLY. No API keys, no real orders, public market data only. THE STRATEGY IN ONE PARAGRAPH Chart 2 (1hr, 0.1% boxes, from closed candles) prints a brick; the START of the newest brick is the LINE. Chart 1 (live trades, 0.1% boxes) must TOUCH the line from the line's side to switch trading ON, and trading stays on while chart 1 still touches it. Chart 3 (live trades, 0.001% boxes) does every entry and exit: a brick that touches the line closes the old trade and opens one in its colour. When chart 1 leaves the line the last trade is HELD (final direction), and it closes when price returns to touch the newest line. That close books the cycle — every trade since the last booking, fees included — and a profit sends 60% out. THE THREE THINGS THAT DECIDE THE RESULT 1. Fills. FILL="line" assumes the flip happens exactly on the line (best case). FILL="market" fills at the last trade price + 1 tick (realistic). 2. Fees. One market order = FEE x position = 0.05% x (cash x 75 x 0.95). On $1 that is $0.0356, so a round trip is $0.0713 — about one full 0.1% brick (7.5%). Neutral flips on the line are what eat a cycle. 3. Holding. Money only comes from runs of several bricks between lines. RUN: pip install python-binance python zec_renko_bot.py ALERTS: terminal + bell always. Phone push needs either NTFY_TOPIC (ntfy.sh, no account and no bot — just pick a private topic name) or TELEGRAM_BOT_TOKEN + TELEGRAM_CHAT_ID. Set them in the environment, never in this file. """ import asyncio import csv import json import os import threading import urllib.parse import urllib.request from datetime import datetime, timezone from decimal import Decimal as D, ROUND_DOWN, ROUND_HALF_UP from binance import AsyncClient, BinanceSocketManager # ── settings ──────────────────────────────────────────────────────────────── SYMBOL = "ZECUSDT" BOX_HOURLY = D("0.001") # chart 2 · 1hr · 0.1% → the line BOX_GATE = D("0.001") # chart 1 · tick · 0.1% → the on/off switch BOX_TRADE = D("0.00001") # chart 3 · tick · 0.001% → every entry and exit LEVERAGE = 75 START_CASH = D("1") # paper USDT SAVINGS_PCT = D("60") # % of a profitable cycle moved out of capital TAKER_FEE = D("0.0005") # 0.05% of position per market order MARGIN_USE = D("0.95") # share of cash used as margin (room for fees) MAINT = D("0.005") # rough maintenance margin for the liquidation sim FILL = "line" # "line" (ideal) or "market" (last price + 1 tick) TRAIL_BRICKS = 0 # 0 = exit at the newest printed line (original rule) # N = the exit trails N bricks behind the newest line, so a run # is not cut off at the first return to the same line GATE_MODE = "touch3" # "touch3" = chart 3's own touch decides (chart 1 removed) # "chart1" = a separate 0.1% chart must touch the line # "band" = price within GATE_BAND of the line GATE_BAND = D("0.001") # 0.1%: the width of one 0.1% box SPREAD = D("0.000085") # 0.0085% of price — measured on Lighter's real ZEC book. # A market order fills at the far touch, so each order pays # half the spread and a flip pays the full spread. MAKER = False # True = resting limit at the line instead of crossing: # the spread cost disappears and you may EARN it, # at the risk of not being filled. FUNDING_PER_HOUR = D("0.000012") # 0.0012%/hour of open notional, charged on the hour # (Lighter settles hourly). 0 = ignore funding. Flat at the # settlement moment pays nothing. TELEGRAM_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") TELEGRAM_CHAT = os.getenv("TELEGRAM_CHAT_ID") NTFY_TOPIC = os.getenv("NTFY_TOPIC") # https://ntfy.sh/ — phone push, no bot needed STATE_FILE, LOG_FILE = "state.json", "events.csv" # ──────────────────────────────────────────────────────────────────────────── assert FILL in ("line", "market") and GATE_MODE in ("touch3", "chart1", "band") class Renko: """Percentage Renko. feed(price) -> [(open, close, dir), ...] one brick at a time.""" def __init__(self, pct): self.pct = pct self.open = self.close = None self.dir = 0 def feed(self, price): if self.close is None: self.open = self.close = price return [] out = [] while True: o, c, d, k = self.open, self.close, self.dir, self.pct if d >= 0 and price >= c + c * k: new = (c, c + c * k, 1) elif d <= 0 and price <= c - c * k: new = (c, c - c * k, -1) elif d == 1 and price <= o - o * k: new = (o, o - o * k, -1) # reversal needs a full box past the open elif d == -1 and price >= o + o * k: new = (o, o + o * k, 1) else: return out self.open, self.close, self.dir = new out.append(new) def now(): return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") def money(x, base): """'+0.0712 USD (+7.12%)' — USD first, percent of the capital it was made on.""" return f"{x:+.4f} USD ({x / base * 100:+.2f}%)" if base else f"{x:+.4f} USD" class Bot: def __init__(self, client=None): self.client = client self.r1h = Renko(BOX_HOURLY) # chart 2 self.r1 = Renko(BOX_GATE) # chart 1 self.r3 = Renko(BOX_TRADE) # chart 3 self.tick = D("0.01") # replaced from exchange info self.qty_step = D("0.001") self.min_notional = D("5") self.line = self.side = None # reversal line and which side price must be on self.last1 = None # newest chart-1 brick (open, close) self.gate = False # chart 3 allowed to trade self.pos = None self.last_price = None self.hour = None self.trail = None # trailing exit level while a trade is carried self.cycle = dict(pnl=D(0), fees=D(0), funding=D(0), spread=D(0), n=0, base=None) self.capital, self.savings, self.dead = START_CASH, D(0), False self.load() # ── state, logging, alerts ────────────────────────────────────────────── def load(self): if os.path.exists(STATE_FILE): s = json.load(open(STATE_FILE)) self.capital, self.savings = D(s["capital"]), D(s["savings"]) def save(self): json.dump({"capital": str(self.capital), "savings": str(self.savings)}, open(STATE_FILE, "w")) def log(self, event, detail="", pnl="", fees=""): print(f"{now()} | {event:<16} | {detail}" + (f" | PnL {pnl}" if pnl else "") + f" | cash {self.capital:.4f} saved {self.savings:.4f}") if fees: print(f"{'':19} fees: {fees}") new = not os.path.exists(LOG_FILE) with open(LOG_FILE, "a", newline="") as f: w = csv.writer(f) if new: w.writerow(["time", "event", "detail", "pnl", "fees", "capital", "savings"]) w.writerow([now(), event, detail, pnl, fees, f"{self.capital:.6f}", f"{self.savings:.6f}"]) def alert(self, title, msg): text = f"{SYMBOL} {title}: {msg}" print(f"\a>>> ALERT {text}") if NTFY_TOPIC: threading.Thread(target=self._ntfy, args=(title, msg), daemon=True).start() if TELEGRAM_TOKEN and TELEGRAM_CHAT: threading.Thread(target=self._telegram, args=(text,), daemon=True).start() def _ntfy(self, title, msg): try: req = urllib.request.Request( f"https://ntfy.sh/{NTFY_TOPIC}", data=msg.encode(), method="POST", headers={"Title": f"{SYMBOL} {title}", "Tags": "chart_with_upwards_trend"}) urllib.request.urlopen(req, timeout=10) except Exception as e: print("ntfy failed:", e) def _telegram(self, text): try: data = urllib.parse.urlencode({"chat_id": TELEGRAM_CHAT, "text": text}).encode() urllib.request.urlopen( f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage", data, timeout=10) except Exception as e: print("telegram failed:", e) # hooks the backtester fills in; no-ops live def on_close(self, net, base, move): pass def on_book(self, total, fees, n, base): pass # ── startup ───────────────────────────────────────────────────────────── async def seed(self): info = await self.client.futures_exchange_info() sym = next((s for s in info["symbols"] if s["symbol"] == SYMBOL), None) if sym is None: raise SystemExit(f"{SYMBOL} not found on Binance USD-M futures") for f in sym["filters"]: if f["filterType"] == "PRICE_FILTER": self.tick = D(f["tickSize"]) elif f["filterType"] == "LOT_SIZE": self.qty_step = D(f["stepSize"]) elif f["filterType"] == "MIN_NOTIONAL": self.min_notional = D(f["notional"]) kl = await self.client.futures_klines(symbol=SYMBOL, interval="1h", limit=1000) ms = int(datetime.now(timezone.utc).timestamp() * 1000) for k in kl: if k[6] < ms: # closed hourly candles only self.r1h.feed(D(k[4])) if self.r1h.dir: self.line, self.side = self.to_tick(self.r1h.open), self.r1h.dir px = D(kl[-1][4]) self.log("SETUP", f"tick {self.tick} | step {self.qty_step} | min {self.min_notional} USDT | " f"ZEC ~{px} | 0.1% box {px * BOX_HOURLY:.3f} | " f"0.001% box {px * BOX_TRADE:.4f} = {px * BOX_TRADE / self.tick:.1f} ticks") self.log("START", f"paper | fill={FILL} | cash {self.capital} USDT | " f"line {self.line} side {'up' if self.side == 1 else 'down'} | tele {bool(TELEGRAM_TOKEN)}") def to_tick(self, x): return (x / self.tick).quantize(D("1"), rounding=ROUND_HALF_UP) * self.tick # ── orders (paper) ────────────────────────────────────────────────────── def open(self, d, raw, slip, on_line): px = raw + d * slip # buy a little higher, sell a little lower qty = (self.capital * MARGIN_USE * D(str(LEVERAGE)) / px / self.qty_step ).to_integral_value(rounding=ROUND_DOWN) * self.qty_step notional = qty * px if notional < self.min_notional: self.log("STOP", f"notional {notional:.2f} < minimum {self.min_notional} USDT") self.dead = True return fee = notional * TAKER_FEE cross = D(0) if MAKER else notional * SPREAD / 2 # a resting limit crosses nothing self.capital -= fee + cross self.cycle["pnl"] -= cross self.cycle["spread"] += cross if self.cycle["base"] is None: self.cycle["base"] = self.capital + fee base = self.cycle["base"] self.pos = dict(dir=d, entry=px, qty=qty, fee=fee, base=base, armed=on_line, side=-d, carried=False, raw=raw) name = "LONG" if d == 1 else "SHORT" self.log(f"OPEN {name}", f"@ {px} | {qty} ZEC | {notional:.2f} USD | line {self.line}", fees=f"entry {fee:.4f} USD ({fee / base * 100:.2f}% of capital)") self.alert(f"{name} opened", f"@ {px} | {notional:.2f} USD notional | " f"fee {fee:.4f} USD ({fee / base * 100:.2f}%)") self.save() def close(self, raw, slip, reason): p = self.pos px = raw - p["dir"] * slip gross = p["dir"] * (px - p["entry"]) * p["qty"] fee_c = px * p["qty"] * TAKER_FEE cross_c = D(0) if MAKER else px * p["qty"] * SPREAD / 2 net = gross - p["fee"] - fee_c - cross_c self.capital += gross - fee_c - cross_c self.cycle["spread"] += cross_c self.cycle["pnl"] += net self.cycle["fees"] += p["fee"] + fee_c self.cycle["n"] += 1 move = p["dir"] * (raw - p["raw"]) # price move only, slippage excluded label = ("3rd chart neutral" if abs(move) <= self.tick / 2 else "3rd chart win" if move > 0 else "3rd chart loss") name = "LONG" if p["dir"] == 1 else "SHORT" base = p["base"] self.pos = None self.log(label, f"CLOSE {name} {p['entry']} -> {px} | {reason}", pnl=money(net, base), fees=f"open {p['fee']:.4f} + close {fee_c:.4f} = " f"{p['fee'] + fee_c:.4f} USD ({(p['fee'] + fee_c) / base * 100:.2f}% of capital)") self.alert(f"{label} · PnL {money(net, base)}", f"{name} {p['entry']} -> {px} | fees {(p['fee'] + fee_c):.4f} USD") self.on_close(net, base, move) self.save() def flip(self, d, touched): carried = bool(self.pos and self.pos["carried"]) raw = self.line if touched else self.last_price slip = D(0) if (touched and FILL == "line") else self.tick if self.pos: self.close(raw, slip, "touched the line" if touched else "printed back without touching") if carried: self.book("held trade closed at the line") if not self.dead: self.open(d, raw, slip, touched) def book(self, reason): c = self.cycle total, fees, n, base = c["pnl"], c["fees"], c["n"], c["base"] or self.capital if total > 0: out = total * SAVINGS_PCT / 100 # never strand the account: withdrawing must leave enough to fund one order floor = self.min_notional / (MARGIN_USE * D(str(LEVERAGE))) if self.capital - out < floor: out = max(D(0), self.capital - floor) self.capital -= out self.savings += out self.log("2nd chart profit", f"{reason} | {n} trades | keep {total - out:.4f} | " f"MOVE TO WALLET {out:.4f} USD", pnl=money(total, base), fees=f"{fees:.4f} fees + {c.get('funding', D(0)):.4f} funding = " f"{fees + c.get('funding', D(0)):.4f} USD over {n} trades") self.alert(f"CYCLE PROFIT {money(total, base)}", f"move {out:.4f} USD ({SAVINGS_PCT}%) to your wallet | fees {fees:.4f} USD") else: self.log("2nd chart loss/flat", reason, pnl=money(total, base), fees=f"{fees:.4f} fees + {c.get('funding', D(0)):.4f} funding = " f"{fees + c.get('funding', D(0)):.4f} USD over {n} trades") self.alert(f"CYCLE {'FLAT' if total == 0 else 'LOSS'} {money(total, base)}", f"nothing to send out | fees {fees:.4f} USD") self.on_book(total, fees, n, base) self.cycle = dict(pnl=D(0), fees=D(0), funding=D(0), spread=D(0), n=0, base=None) self.save() def accrue_funding(self, price): """One hour of funding on whatever is open right now — nothing if we are flat.""" if not self.pos or FUNDING_PER_HOUR <= 0: return notional = self.pos["qty"] * price fee = notional * FUNDING_PER_HOUR self.capital -= fee self.cycle["pnl"] -= fee # rides inside the P&L, reported on its own line self.cycle["funding"] += fee self.log("FUNDING", f"{fee:.6f} USD on {notional:.2f} USD open at {price}") def liquidated(self, price): p = self.pos if p["dir"] * (p["entry"] - price) / p["entry"] >= D(1) / D(str(LEVERAGE)) - MAINT: self.log("LIQUIDATED", f"price {price} vs entry {p['entry']}", pnl=money(-self.capital, p["base"])) self.alert("LIQUIDATED", f"at {price}. Capital gone.") self.capital, self.pos, self.dead = D(0), None, True self.save() return True return False # ── the line and the gate ─────────────────────────────────────────────── def set_line(self, line, side): self.line, self.side = self.to_tick(line), side self.log("NEW LINE", f"{self.line} | price must be {'above' if side == 1 else 'below'} it") self._update_trail() self.check_gate() def _update_trail(self): """The exit ratchets with the run: TRAIL_BRICKS behind the newest line.""" if not (self.pos and self.pos["carried"]): return if TRAIL_BRICKS <= 0: self.trail = None return cand = self.line - self.pos["dir"] * self.line * BOX_HOURLY * TRAIL_BRICKS if self.trail is None: self.trail = cand elif self.pos["dir"] == 1: self.trail = max(self.trail, cand) # long: trail up only else: self.trail = min(self.trail, cand) # short: trail down only def check_gate(self): if self.line is None: return L = self.line if GATE_MODE == "band": px = self.last_price if px is None: return self._gate(abs(px - L) <= L * GATE_BAND, (px >= L) if self.side == 1 else (px <= L), "price") elif GATE_MODE == "chart1": if self.last1 is None: return o, c = self.last1 self._gate(min(o, c) <= L <= max(o, c), (c >= L) if self.side == 1 else (c <= L), "chart 1") def _gate(self, inside, ahead, who): """inside = something is touching the line, ahead = price is on the line's side.""" if self.gate and not inside: self.gate = False self.trail = None if self.pos: # final direction decided: hold it self.pos["carried"], self.pos["armed"] = True, False name = "LONG" if self.pos["dir"] == 1 else "SHORT" self.log("HOLD", f"final direction {name} @ {self.pos['entry']} | " f"held until price touches a line") self.alert(f"{name} HELD", f"final direction from {self.pos['entry']} — " f"closes when price touches a line") elif not self.gate and inside and ahead: self.gate = True self.log("GATE ON", f"{who} at line {self.line}: chart 3 trades") self.alert("TRADING ON", f"{who} touched the line at {self.line}") def on_chart1(self, o, c): self.last1 = (o, c) self.check_gate() def on_chart3(self, o, c, d): if self.line is None or self.dead: return L, p = self.line, self.pos lo, hi = min(o, c), max(o, c) touches = lo <= L <= hi if p and p["carried"]: # held trade: the line or its trail is the ONLY exit Lx = self.trail if self.trail is not None else L if lo <= Lx <= hi: raw = Lx if FILL == "line" else self.last_price self.close(raw, D(0) if FILL == "line" else self.tick, "held trade closed at the line") self.book("held trade closed at the line") return if GATE_MODE == "touch3": self._gate(touches, (c >= L) if self.side == 1 else (c <= L), "chart 3") if not self.gate: return if touches: # the brick touches the line if p is None or p["dir"] != d: self.flip(d, touched=True) else: p["armed"], p["side"] = True, -d elif p and p["armed"] and d != p["dir"] and ((1 if lo > L else -1) == p["side"]): self.flip(d, touched=False) # printed back without touching: one-brick loss # ── live feed ─────────────────────────────────────────────────────────── async def new_hour(self): kl = await self.client.futures_klines(symbol=SYMBOL, interval="1h", limit=3) ms = int(datetime.now(timezone.utc).timestamp() * 1000) closed = [k for k in kl if k[6] < ms] if not closed: return bricks = self.r1h.feed(D(closed[-1][4])) for b in bricks: # EVERY brick is its own line — an hour can self.set_line(b[0], b[2]) # print 7 or 8 of them, and the newest one wins async def on_trade(self, price, ts): self.last_price = price if self.pos and self.liquidated(price): return hour = ts // 3_600_000 if self.hour is None: self.hour = hour elif hour != self.hour: self.hour = hour self.accrue_funding(price) # the hour that just closed is now settled await self.new_hour() for o, c, _ in self.r1.feed(price): self.on_chart1(o, c) for o, c, d in self.r3.feed(price): self.on_chart3(o, c, d) async def main(): client = await AsyncClient.create() bot = Bot(client) await bot.seed() sm = BinanceSocketManager(client) async with sm.aggtrade_socket(SYMBOL) as stream: while True: m = await stream.recv() await bot.on_trade(D(m["p"]), m["T"]) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: print("\nstopped")