""" Dashboard Flask live multi-asset - AI Trading Engine (v3 lightweight) Server leggero: candele live da Bybit (ZEC + AERO + DASH + SOL) + segnali + equity + posizioni. 5 route: / → vista riassuntiva (grid 2x2 con tutti i 4 setup) /zec → full-screen ZEC 4H /aero → full-screen AERO 4H /dash → full-screen DASH 4H /sol → full-screen SOL 4H /api/data → JSON aggregato Nessun grafico pesante (no Plotly/Chart.js). Solo tabelle + candele ASCII-style. Read-only, no auth richiesta. """ import os import json from pathlib import Path from datetime import datetime, timezone import ccxt import pandas as pd import numpy as np from flask import Flask, jsonify, render_template_string, make_response ENV_FILE = Path(r"G:\AI TRADING ENGINE\Bybit_new.env") STATE_DIR = Path(__file__).parent SETUPS = [ {"id": "zec", "name": "ZEC_4H", "symbol": "ZEC/USDT:USDT", "tf": "4h", "mask": "10001011", "bb_length": 55, "bb_mult": 1.0, "roc_length": 36, "atr_length": 16, "margin": 500, "validation": "🟢 TRIPLA VALIDATA"}, {"id": "aero", "name": "AERO_4H", "symbol": "AERO/USDT:USDT", "tf": "4h", "mask": "01010101", "bb_length": 55, "bb_mult": 1.0, "roc_length": 36, "atr_length": 16, "margin": 250, "validation": "🟡 2/3 BORDERLINE"}, {"id": "dash", "name": "DASH_4H", "symbol": "DASH/USDT:USDT", "tf": "4h", "mask": "10000111", "bb_length": 55, "bb_mult": 1.0, "roc_length": 36, "atr_length": 16, "margin": 250, "validation": "🟡 2/3 BORDERLINE"}, {"id": "pyth_dls", "name": "PYTH_3H_CLL_CLS_CSL_DLS", "symbol": "PYTH/USDT:USDT", "tf": "3h", "mask": "00100111", "bb_length": 55, "bb_mult": 1.0, "roc_length": 36, "atr_length": 16, "margin": 250, "validation": "🆕 PYTH 00100111"}, {"id": "pendle_cdl", "name": "PENDLE_4H_CSL_DLL_DLS_DSL","symbol": "PENDLE/USDT:USDT", "tf": "4h", "mask": "00101110", "bb_length": 55, "bb_mult": 1.0, "roc_length": 36, "atr_length": 16, "margin": 250, "validation": "🆕 PENDLE 00101110"}, {"id": "sol", "name": "SOL_4H_NAIVE", "symbol": "SOL/USDT:USDT", "tf": "4h", "mask": "00101000", "bb_length": 55, "bb_mult": 1.0, "roc_length": 36, "atr_length": 16, "margin": 0, "validation": "📡 NAIVE ALERT (no Charter)"}, ] app = Flask(__name__) _cache = {} _cache_ttl = 60 # secondi def get_client(): return ccxt.bybit({"enableRateLimit": True, "options": {"defaultType": "linear"}}) def compute_indicators(df, bb_len, factor, roc_len, atr_len): df = df.copy() df.columns = [c.strip().lower() for c in df.columns] close = df["close"].astype(float); high = df["high"].astype(float); low = df["low"].astype(float) vol = df["volume"].astype(float) sma = close.rolling(bb_len, min_periods=bb_len).mean() sd = close.rolling(bb_len, min_periods=bb_len).std(ddof=0) df["bb_plus"] = sma + factor * sd df["bb_minus"] = sma - factor * sd prev = close.shift(roc_len) df["roc"] = (close - prev) / prev * 100.0 tr = pd.concat([high - low, (high - low.shift(1)).abs(), (low - high.shift(1)).abs()], axis=1).max(axis=1) df["atr"] = tr.ewm(alpha=1.0/atr_len, adjust=False).mean() sp = np.where(close > df["open"].values, 1, np.where(close < df["open"].values, -1, 0)) sv = np.where(vol > np.roll(vol.values, 1), 1, np.where(vol < np.roll(vol.values, 1), -1, 0)) sr = np.where(df["roc"].values > 0, 1, np.where(df["roc"].values < 0, -1, 0)) cd_idx = np.full(len(df), -1, dtype=np.int8) valid = (sp != 0) & (sv != 0) & (sr != 0) & ~np.isnan(df["roc"].values) & ~np.isnan(df["bb_plus"].values) m1 = valid & (sp == 1) & (sv == 1); cd_idx[m1 & (sr > 0)] = 0; cd_idx[m1 & (sr < 0)] = 1 m2 = valid & (sp == -1) & (sv == -1); cd_idx[m2 & (sr > 0)] = 2; cd_idx[m2 & (sr < 0)] = 3 m3 = valid & (sp == 1) & (sv == -1); cd_idx[m3 & (sr > 0)] = 4; cd_idx[m3 & (sr < 0)] = 5 m4 = valid & (sp == -1) & (sv == 1); cd_idx[m4 & (sr > 0)] = 6; cd_idx[m4 & (sr < 0)] = 7 df["cd_idx"] = cd_idx return df def fetch_setup_cached(ex, setup): key = setup["id"] now = datetime.now().timestamp() if key in _cache and (now - _cache[key]["ts"]) < _cache_ttl: return _cache[key]["data"] ohlcv = ex.fetch_ohlcv(setup["symbol"], setup["tf"], limit=200) df = pd.DataFrame(ohlcv, columns=["timestamp", "open", "high", "low", "close", "volume"]) df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True) df = compute_indicators(df, setup["bb_length"], setup["bb_mult"], setup["roc_length"], setup["atr_length"]) _cache[key] = {"ts": now, "data": df} return df def detect_signal(df, mask_str): if len(df) < 100: return None last = df.iloc[-1] if pd.isna(last["bb_plus"]) or pd.isna(last["roc"]): return None mask_padded = mask_str.zfill(8) cd_active = [int(c) for c in mask_padded] cd = int(last["cd_idx"]) if cd < 0 or not cd_active[cd]: return None close = float(last["close"]); atr = float(last["atr"]) if close > float(last["bb_plus"]) and last["roc"] > 0: return "LONG" if close < float(last["bb_minus"]) and last["roc"] < 0: return "SHORT" return None def make_sparkline(values, width=20, height=20): """Crea sparkline ASCII-style (no JS, no charts).""" if len(values) < 2: return "..." mn, mx = min(values), max(values) if mx == mn: return "─" * width bars = "▁▂▃▄▅▆▇█" step = (mx - mn) / (len(bars) - 1) line = "" for v in values[-width:]: idx = min(int((v - mn) / step), len(bars) - 1) line += bars[idx] return line def fetch_setup_full(ex, setup): try: df = fetch_setup_cached(ex, setup) sig = detect_signal(df, setup["mask"]) last = df.iloc[-1] # Stats df_30 = df.tail(180) n = len(df_30) vol_30 = df_30["close"].pct_change().std() * np.sqrt(365 * 6) * 100 high_30 = df_30["high"].max() low_30 = df_30["low"].min() range_30_pct = (high_30 - low_30) / low_30 * 100 # Candele ultime 12 per tabella candles = [] for _, row in df.tail(12).iterrows(): cd = int(row["cd_idx"]) cd_name = ["CLL", "CLS", "CSL", "CSS", "DLL", "DLS", "DSL", "DSS"][cd] if cd >= 0 else "N/A" candles.append({ "ts": row["timestamp"].strftime("%d/%m %H:%M"), "close": float(row["close"]), "bb_plus": float(row["bb_plus"]) if not pd.isna(row["bb_plus"]) else 0, "bb_minus": float(row["bb_minus"]) if not pd.isna(row["bb_minus"]) else 0, "roc": float(row["roc"]) if not pd.isna(row["roc"]) else 0, "atr": float(row["atr"]) if not pd.isna(row["atr"]) else 0, "cd_name": cd_name, }) # Sparkline close sparkline = make_sparkline(df["close"].tail(50).tolist(), width=30) return { "id": setup["id"], "name": setup["name"], "symbol": setup["symbol"], "tf": setup["tf"], "mask": setup["mask"], "validation": setup["validation"], "last_close": float(last["close"]), "last_ts": last["timestamp"].isoformat(), "atr": float(last["atr"]) if not pd.isna(last["atr"]) else 0, "bb_plus": float(last["bb_plus"]) if not pd.isna(last["bb_plus"]) else 0, "bb_minus": float(last["bb_minus"]) if not pd.isna(last["bb_minus"]) else 0, "roc": float(last["roc"]) if not pd.isna(last["roc"]) else 0, "signal": sig, "n_candles": int(n), "vol_30d_pct": round(vol_30, 2), "high_30d": float(high_30), "low_30d": float(low_30), "range_30d_pct": round(range_30_pct, 2), "sparkline": sparkline, "candles": candles, } except Exception as e: return {"id": setup["id"], "name": setup["name"], "error": str(e)} DASHBOARD_HTML = """ AI Trading Engine — Live Dashboard

AI Trading Engine — Live Dashboard

LIVE Bybit Demo
📊 ALL (5 setup) 🟢 ZEC 4H 🟡 AERO 4H 🟡 DASH 4H 🆕 PYTH 3H 00100111 🆕 PENDLE 4H CSL+DLL+DLS+DSL 📡 SOL 4H

⏳ Caricamento da Bybit...

-- · Bybit Demo (api-demo.bybit.com) · Read-only · Charter: TP1+3% / TP2+5% / SL ATR 2x · Live bot Charter attivo · Refresh ogni 60s
""" def render(filter_id=""): response = make_response(render_template_string(DASHBOARD_HTML.replace("__ACTIVE_FILTER__", filter_id))) response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0" response.headers["Pragma"] = "no-cache" response.headers["Expires"] = "0" return response @app.route("/") def index(): return render("") @app.route("/") def asset_page(asset_id): valid = [s["id"] for s in SETUPS] if asset_id not in valid: return render(""), 404 return render(asset_id) @app.route("/api/all") def api_all(): try: ex = get_client() return jsonify({"setups": [fetch_setup_full(ex, s) for s in SETUPS]}) except Exception as e: return jsonify({"error": str(e), "setups": []}), 500 @app.route("/api/") def api_one(asset_id): setup = next((s for s in SETUPS if s["id"] == asset_id), None) if not setup: return jsonify({"error": "not found", "setups": []}), 404 try: ex = get_client() return jsonify({"setups": [fetch_setup_full(ex, setup)]}) except Exception as e: return jsonify({"error": str(e), "setups": []}), 500 if __name__ == "__main__": app.run(host="127.0.0.1", port=5001, debug=False)