"""
Monitor visuale INTRADAY 30m per la strategia RETTANGOLO (breakout + retest).
- Range: high/low candela daily precedente (rettangolo blu)
- Candele: 2H intraday (ultimi 7 giorni)
- Segnale live: candela 2H corrente, se c'è breakout precedente nelle ultime 10 candele
e la corrente tocca il livello rotto con pattern di inversione -> entry
- Auto-refresh: rigenera ogni 60s
"""
import sys
import time
import json
import urllib.request
import urllib.parse
from datetime import datetime, timezone
from pathlib import Path
from zoneinfo import ZoneInfo
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from rettangolo_strategy import compute_signal, scan_signals
from rettangolo_config import load_symbols
# Fuso orario di default per visualizzazione candele: Europe/Rome (CEST/CET auto)
LOCAL_TZ = ZoneInfo("Europe/Rome")
OUT_HTML = Path(r"G:\AI TRADING ENGINE\live_deploy\bt_rettangolo\monitor.html")
ASSETS = load_symbols() # letti da rettangolo_assets.csv (centralizzato)
def fetch_klines(symbol, interval, days):
end_ms = int(time.time() * 1000)
start_ms = end_ms - days * 24 * 3600 * 1000
params = {"category": "linear", "symbol": symbol, "interval": interval, "start": start_ms, "end": end_ms, "limit": 200}
qs = urllib.parse.urlencode(params)
url = f"https://api.bybit.com/v5/market/kline?{qs}"
with urllib.request.urlopen(url, timeout=15) as r:
data = json.loads(r.read().decode("utf-8"))
rows = data.get("result", {}).get("list", [])
rows.sort(key=lambda x: int(x[0]))
klines = []
for r in rows:
klines.append({
"ts": int(r[0]),
"date": datetime.fromtimestamp(int(r[0])/1000, tz=LOCAL_TZ).strftime("%Y-%m-%d %H:%M"),
"open": float(r[1]),
"high": float(r[2]),
"low": float(r[3]),
"close": float(r[4]),
"volume": float(r[5])
})
return klines
def load_real_trades(symbol, n=5):
"""Carica gli ultimi N trade REALI eseguiti su Bybit per il symbol specifico.
Ritorna lista di dict con: ts, ts_str, side, entry_price, exit_price, pnl_pct, reason.
Source: webhook_listener/logs/trades.csv (generato da webhook_receiver.py)."""
import csv as _csv
trades_csv = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_listener\logs\trades.csv")
if not trades_csv.exists():
return []
rows = []
try:
with open(trades_csv, "r", encoding="utf-8") as f:
reader = _csv.DictReader(f)
for row in reader:
if row.get("symbol") != symbol:
continue
try:
ts_iso = row.get("timestamp", "")
ts = datetime.fromisoformat(ts_iso).timestamp() * 1000
entry_str = row.get("entry_price", "").strip()
exit_str = row.get("exit_price", "").strip()
pnl_str = row.get("pnl_pct", "").strip()
rows.append({
"ts": ts,
"ts_str": datetime.fromtimestamp(ts/1000, tz=LOCAL_TZ).strftime("%Y-%m-%d %H:%M"),
"side": row.get("side", "Buy"),
"entry_price": float(entry_str) if entry_str else None,
"exit_price": float(exit_str) if exit_str else None,
"pnl_pct": float(pnl_str) if pnl_str else None,
"reason": row.get("reason", ""),
})
except Exception:
continue
rows.sort(key=lambda r: r["ts"], reverse=True)
return rows[:n]
except Exception:
return []
def detect_position_strategy(symbol, side, created_unix_ms):
"""Ritorna la strategy che ha aperto la posizione: 'vptr3' | 'rettangolo' | 'audit' | 'unknown'.
Match tramite orders.log (vptr3/audit) e rettangolo_runner.log (rettangolo)."""
# 1. Cerca in orders.log (VPTR3 / audit_test / manual) - case insensitive
orders_log = Path(r"G:\AI TRADING ENGINE\live_deploy\logs\orders.log")
if orders_log.exists():
try:
with open(orders_log, "r", encoding="utf-8") as f:
for line in f:
line_l = line.lower()
if symbol.lower() not in line_l or side.lower() not in line_l:
continue
if "strategy=vptr3" in line_l:
return "vptr3"
if "strategy=audit_test" in line_l:
return "audit"
except Exception:
pass
# 2. Cerca in rettangolo_runner.log (RETTANGOLO)
rett_log = Path(r"G:\AI TRADING ENGINE\live_deploy\webhook_listener\logs\rettangolo_runner.log")
if rett_log.exists():
try:
# createdTime e' in ms, timestamp del log e' ISO locale Europe/Rome
from datetime import datetime
from zoneinfo import ZoneInfo
tz = ZoneInfo("Europe/Rome")
target_local = datetime.fromtimestamp(created_unix_ms / 1000, tz=tz)
target_str_a = target_local.strftime("%Y-%m-%dT%H:%M")
target_str_b = target_local.strftime("%Y-%m-%dT%H:%M:%S")
with open(rett_log, "r", encoding="utf-8") as f:
for line in f:
if "ORDER APERTA" not in line or symbol not in line or side.capitalize() not in line:
continue
if target_str_a in line or target_str_b in line:
return "rettangolo"
except Exception:
pass
return "unknown"
def build_figure(symbol, intraday_klines, daily_klines):
if len(intraday_klines) < 5 or len(daily_klines) < 2:
return None
# FIX 2026-07-17 (Verifier): filtro per SESSIONE LONDON 9-9 (non Bybit daily K-line mezzanotte UTC)
cur_day = daily_klines[-1]
prev_day = daily_klines[-2]
# Calcola finestra sessione London 9-9
_now_london = datetime.now(ZoneInfo("Europe/London"))
_session_open_today_london = _now_london.replace(hour=9, minute=0, second=0, microsecond=0)
_session_open_today_unix_ms = int(_session_open_today_london.timestamp()) * 1000
_session_open_yesterday_unix_ms = _session_open_today_unix_ms - 86400 * 1000
filtered = [k for k in intraday_klines if _session_open_yesterday_unix_ms <= k["ts"] <= _session_open_today_unix_ms]
if len(filtered) < 3:
filtered = intraday_klines # fallback se filtro troppo aggressivo
# Box CUSTOM aggregato da candele 2H filtrate (NON Bybit daily K-line)
if filtered:
custom_high = max(k["high"] for k in filtered)
custom_low = min(k["low"] for k in filtered)
custom_date = datetime.fromtimestamp(_session_open_yesterday_unix_ms / 1000, tz=LOCAL_TZ).strftime("%Y-%m-%d")
else:
custom_high = prev_day["high"]
custom_low = prev_day["low"]
custom_date = prev_day["date"][:10]
prev_daily = {"date": custom_date, "ts": _session_open_yesterday_unix_ms,
"open": custom_low, "high": custom_high, "low": custom_low, "close": custom_low}
# RANGE dal box custom (24h London 9-9)
rng_top = custom_high
rng_bot = custom_low
rng_mid = (rng_top + rng_bot) / 2
rng_pct = (rng_top - rng_bot) / rng_bot * 100
# candele 2H (solo ieri + oggi)
dates = [k["date"] for k in filtered]
opens = [k["open"] for k in filtered]
highs = [k["high"] for k in filtered]
lows = [k["low"] for k in filtered]
closes = [k["close"] for k in filtered]
# FIX 2026-07-17 (Verifier): split a 9:00 London oggi (= 10:00 Italy in estate), non mezzanotte UTC
idx_today = next((i for i, k in enumerate(filtered) if k["ts"] >= _session_open_today_unix_ms), len(filtered))
yesterday = filtered[:idx_today]
today = filtered[idx_today:]
# Sep date per il separatore verticale
sep_date = datetime.fromtimestamp(_session_open_today_unix_ms / 1000, tz=LOCAL_TZ).strftime("%Y-%m-%d %H:%M")
# segnale live
sig = compute_signal(prev_daily, filtered, len(filtered) - 1)
sig_text = "NESSUN SETUP - in attesa breakout + retest"
sig_color = "gray"
if sig:
sig_text = ">>> " + sig["signal"] + " ENTRY @ " + format(sig["entry"], ".4f") + " | SL=" + format(sig["sl"], ".4f") + " | TP=" + format(sig["tp"], ".4f") + " <<<"
sig_color = "#4caf50" if sig["signal"] == "LONG" else "#ef5350"
# label DIREZIONE compatta per chart (paper coords, in alto a destra)
if sig:
dir_text = sig["signal"] + " @" + format(sig["entry"], ".4f")
dir_color = "#4caf50" if sig["signal"] == "LONG" else "#ef5350"
dir_bg = "rgba(76,175,80,0.15)" if sig["signal"] == "LONG" else "rgba(239,83,80,0.15)"
else:
dir_text = "NESSUN SETUP"
dir_color = "#888"
dir_bg = "rgba(136,136,136,0.10)"
historical = scan_signals(prev_daily, filtered)
historical = historical[-5:]
# figura
fig = make_subplots(rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.05, row_heights=[0.75, 0.25])
# ieri: candele piene, colori saturi
fig.add_trace(go.Candlestick(
x=[k["date"] for k in yesterday], open=[k["open"] for k in yesterday],
high=[k["high"] for k in yesterday], low=[k["low"] for k in yesterday],
close=[k["close"] for k in yesterday],
name=symbol + " 2H IERI (chiuso)", increasing_line_color="#26a69a", decreasing_line_color="#ef5350",
line=dict(width=1.5)
), row=1, col=1)
# oggi: candele in formazione - colori pastello PIENI (no opacity) + outline 2px
# niente opacity=0.5 sul trace: rende il fill trasparente e l'outline "dashed-style" su dark bg
fig.add_trace(go.Candlestick(
x=[k["date"] for k in today], open=[k["open"] for k in today],
high=[k["high"] for k in today], low=[k["low"] for k in today],
close=[k["close"] for k in today],
name=symbol + " 2H OGGI (live)", increasing_line_color="#4dd0c1", decreasing_line_color="#e57373",
line=dict(width=2)
), row=1, col=1)
# sfondo leggermente più scuro sulla zona "oggi" per separazione visiva
if today:
today_first = today[0]["date"]
today_last = today[-1]["date"]
# estendi la zona di 2h a sinistra (fino al confine di ieri) e 2h a destra
fig.add_shape(
type="rect",
x0=today_first, x1=today_last,
y0=rng_bot - (rng_top - rng_bot) * 0.5, y1=rng_top + (rng_top - rng_bot) * 0.5,
line=dict(width=0),
fillcolor="rgba(255,255,255,0.04)",
layer="below"
)
# FIX 2026-07-17 (Verifier): separatore verticale a 9:00 London (= 10:00 Italy estate), non mezzanotte UTC
if today and yesterday:
sep_x = sep_date
fig.add_shape(
type="line", x0=sep_x, x1=sep_x,
y0=rng_bot - (rng_top - rng_bot) * 0.5, y1=rng_top + (rng_top - rng_bot) * 0.5,
line=dict(color="rgba(255,255,255,0.65)", width=1.5, dash="dash")
)
# label "OGGI (live)" in alto a destra (paper coords su ENTRAMBI gli assi) - non overlappa con candele
fig.add_annotation(
xref="paper", yref="paper", x=0.95, y=1.02,
text="OGGI (live) →", showarrow=False,
xanchor="right", yanchor="bottom",
font=dict(color="#4dd0c1", size=12, family="monospace"),
bgcolor="rgba(77,208,193,0.15)", bordercolor="#4dd0c1", borderwidth=1, borderpad=4
)
# label DIREZIONE (SHORT/LONG/NESSUN SETUP) di fianco a OGGI, leggermente distanziata
fig.add_annotation(
xref="paper", yref="paper", x=0.71, y=1.02,
text="" + dir_text + "", showarrow=False,
xanchor="right", yanchor="bottom",
font=dict(color=dir_color, size=12, family="monospace"),
bgcolor=dir_bg, bordercolor=dir_color, borderwidth=1, borderpad=4
)
# FIX 2026-07-18 (Verifier): rettangolo split a 9:00 London - ieri linea continua, oggi tratteggiato
fig.add_shape(
type="rect", x0=dates[0], x1=sep_date, y0=rng_bot, y1=rng_top,
line=dict(color="rgba(33,150,243,0.7)", width=1.5), fillcolor="rgba(33,150,243,0.06)",
name="Range " + prev_daily["date"][:10] + " (IERI - chiuso)"
)
fig.add_shape(
type="rect", x0=sep_date, x1=dates[-1], y0=rng_bot, y1=rng_top,
line=dict(color="rgba(33,150,243,0.7)", width=1.5, dash="dash"), fillcolor="rgba(33,150,243,0.03)",
name="Range " + prev_daily["date"][:10] + " (OGGI - live, in formazione)"
)
# LINEE TOP/BOTTOM: ieri continue, oggi tratteggiate (split a 9:00 London)
fig.add_shape(
type="line", x0=dates[0], x1=sep_date, y0=rng_top, y1=rng_top,
line=dict(color="rgba(33,150,243,1.0)", width=2.5),
name="TOP " + format(rng_top, ".4f") + " (IERI)"
)
fig.add_shape(
type="line", x0=sep_date, x1=dates[-1], y0=rng_top, y1=rng_top,
line=dict(color="rgba(33,150,243,1.0)", width=2.5, dash="dash"),
name="TOP " + format(rng_top, ".4f") + " (OGGI)"
)
fig.add_shape(
type="line", x0=dates[0], x1=sep_date, y0=rng_bot, y1=rng_bot,
line=dict(color="rgba(33,150,243,1.0)", width=2.5),
name="BOTTOM " + format(rng_bot, ".4f") + " (IERI)"
)
fig.add_shape(
type="line", x0=sep_date, x1=dates[-1], y0=rng_bot, y1=rng_bot,
line=dict(color="rgba(33,150,243,1.0)", width=2.5, dash="dash"),
name="BOTTOM " + format(rng_bot, ".4f") + " (OGGI)"
)
# linea MID (target TP): ieri continua, oggi tratteggiata
fig.add_shape(
type="line", x0=dates[0], x1=sep_date, y0=rng_mid, y1=rng_mid,
line=dict(color="rgba(255,152,0,1.0)", width=2),
name="MID " + format(rng_mid, ".4f") + " (IERI)"
)
fig.add_shape(
type="line", x0=sep_date, x1=dates[-1], y0=rng_mid, y1=rng_mid,
line=dict(color="rgba(255,152,0,1.0)", width=2, dash="dot"),
name="MID " + format(rng_mid, ".4f") + " (OGGI)"
)
for k in filtered:
if k["close"] > rng_top:
fig.add_trace(go.Scatter(
x=[k["date"]], y=[k["close"]], mode="markers",
marker=dict(symbol="triangle-up", size=8, color="#4caf50", opacity=0.6),
showlegend=False, hoverinfo="skip"
), row=1, col=1)
elif k["close"] < rng_bot:
fig.add_trace(go.Scatter(
x=[k["date"]], y=[k["close"]], mode="markers",
marker=dict(symbol="triangle-down", size=8, color="#ef5350", opacity=0.6),
showlegend=False, hoverinfo="skip"
), row=1, col=1)
# etichette prezzi lato destro
fig.add_annotation(
xref="paper", x=0.985, y=rng_top, text="SELL " + format(rng_top, ".4f"),
showarrow=False, xanchor="left", yanchor="middle",
font=dict(color="#ef5350", size=14, family="monospace"),
bgcolor="rgba(239,83,80,0.15)", bordercolor="#ef5350", borderwidth=1, borderpad=4
)
fig.add_annotation(
xref="paper", x=0.985, y=rng_bot, text="BUY " + format(rng_bot, ".4f"),
showarrow=False, xanchor="left", yanchor="middle",
font=dict(color="#4caf50", size=14, family="monospace"),
bgcolor="rgba(76,175,80,0.15)", bordercolor="#4caf50", borderwidth=1, borderpad=4
)
fig.add_annotation(
xref="paper", x=0.985, y=rng_mid, text="TP " + format(rng_mid, ".4f"),
showarrow=False, xanchor="left", yanchor="middle",
font=dict(color="#ff9800", size=14, family="monospace"),
bgcolor="rgba(255,152,0,0.15)", bordercolor="#ff9800", borderwidth=1, borderpad=4
)
for s in historical:
color = "#4caf50" if s["signal"] == "LONG" else "#ef5350"
symbol_arrow = "triangle-up" if s["signal"] == "LONG" else "triangle-down"
fig.add_trace(go.Scatter(
x=[s["date"]], y=[s["entry"]], mode="markers",
marker=dict(symbol=symbol_arrow, size=12, color=color, line=dict(width=2, color="white")),
showlegend=False, hoverinfo="text",
hovertext=s["signal"] + " @ " + format(s["entry"], ".4f")
), row=1, col=1)
colors = ["#26a69a" if closes[i] >= opens[i] else "#ef5350" for i in range(len(closes))]
fig.add_trace(go.Bar(x=dates, y=[k["volume"] for k in filtered], marker_color=colors, name="Volume", showlegend=False), row=2, col=1)
# TRADE PASSATI: triangolino alla data di entry (verde=win, rosso=loss)
# Tolto le linee/label PnL per grafico piu' pulito
real_trades = load_real_trades(symbol, n=5)
for t in real_trades:
if t.get("reason") != "close" or t.get("exit_price") is None:
continue
ts_str = t["ts_str"]
pnl = t.get("pnl_pct", 0) or 0
side = t.get("side", "Buy")
is_long = side.lower() in ("buy", "long")
is_win = pnl > 0
# triangolino entry: WIN verde scuro, LOSS rosso scuro
tri_color = "#2e7d32" if is_win else "#c62828"
tri_symbol = "triangle-up" if is_long else "triangle-down"
# piu' piccolo e opaco di quelli delle posizioni aperte
fig.add_trace(go.Scatter(
x=[ts_str], y=[t["entry_price"]],
mode="markers",
marker=dict(symbol=tri_symbol, size=8, color=tri_color,
opacity=0.5, line=dict(color=tri_color, width=0.5)),
name="Trade",
showlegend=False, hoverinfo="text",
hovertext=f"CHIUSO {side} @ {t['entry_price']:.4f} -> {t['exit_price']:.4f} ({pnl:+.2f}%)"
), row=1, col=1)
# POSIZIONI ATTUALMENTE APERTE SU BYBIT (marker con strategy detection)
# PINE (VPTR3) = quadrato blu, RETT = triangolo verde/rosso, OLD = triangolo grigio
try:
from bybit_demo_client import BybitDemoClient
_bdc = BybitDemoClient()
_open_pos = _bdc.fetch_positions(symbol)
if _open_pos:
_now_str = datetime.now().astimezone().isoformat()
for _p in _open_pos:
_avg = float(_p.get("avgPrice", 0) or 0)
_side_raw = _p.get("side", "")
_created_ms = int(_p.get("createdTime", 0) or 0)
if _avg <= 0:
continue
_is_long = _side_raw.lower() in ("buy", "long")
_strategy = detect_position_strategy(symbol, _side_raw, _created_ms)
# Marker shape + color per strategy
if _strategy == "vptr3":
_marker_symbol = "square"
_marker_color = "#2196f3" # blu PINE
_label_text = "PINE"
elif _strategy == "rettangolo":
_marker_symbol = "triangle-up" if _is_long else "triangle-down"
_marker_color = "#3fb950" if _is_long else "#ef5350"
_label_text = "RETT"
elif _strategy == "audit":
_marker_symbol = "diamond"
_marker_color = "#9c27b0" # viola audit
_label_text = "AUDIT"
else: # unknown / old
_marker_symbol = "triangle-up" if _is_long else "triangle-down"
_marker_color = "#888888" # grigio OLD
_label_text = "OLD"
fig.add_trace(go.Scatter(
x=[_now_str], y=[_avg],
mode="markers",
marker=dict(symbol=_marker_symbol, size=14, color=_marker_color,
line=dict(color="white", width=1.5)),
name="Open",
showlegend=False, hoverinfo="text",
hovertext=f"APERTA {_strategy.upper()} {_side_raw} {_p.get('size','')} @ {_avg}"
), row=1, col=1)
# label "OPEN + strategy" accanto al marker
_label_y = _avg + (rng_top - rng_bot) * 0.04 if _is_long else _avg - (rng_top - rng_bot) * 0.04
fig.add_annotation(
x=_now_str, y=_label_y,
text=f"OPEN {_label_text} {_p.get('size','')} @ {_avg}",
showarrow=False, xanchor="left", yanchor="bottom" if _is_long else "top",
font=dict(color=_marker_color, size=10, family="monospace"),
bgcolor="rgba(0,0,0,0.6)", borderpad=2
)
except Exception:
pass
fig.update_layout(
title="" + symbol + " • Range " + prev_daily["date"][:10] + ": " + format(rng_bot, ".4f") + " — " + format(rng_top, ".4f") + " (" + format(rng_pct, ".2f") + "%) "
"" + sig_text + "",
template="plotly_dark", height=700, width=1500,
xaxis=dict(dtick=3600000, tickformat="%H:%M\n%b %d"),
xaxis2=dict(dtick=3600000, tickformat="%H:%M\n%b %d"),
xaxis_rangeslider_visible=False,
xaxis2_rangeslider_visible=False,
# legenda spostata in basso sotto il volume per non sovrapporsi alle label DIREZIONE/OGGI
legend=dict(orientation="h", y=-0.15, x=0.5, xanchor="center", font=dict(size=10)),
margin=dict(t=80, b=80),
)
return fig
def build_html(per_symbol_data, generated_at, open_symbols=None):
figs = {}
for sym, data in per_symbol_data.items():
fig = build_figure(sym, data["intraday"], data["daily"])
if fig:
figs[sym] = fig
if not figs:
return "
Nessun dato disponibile
"
# Riordina le tab: prima i symbol con posizione aperta (Mattia: in evidenza)
if open_symbols:
ordered = sorted(figs.items(), key=lambda kv: 0 if kv[0] in open_symbols else 1)
else:
ordered = list(figs.items())
tabs_html = ""
divs_html = ""
for i, (sym, fig) in enumerate(ordered):
active = " active" if i == 0 else ""
tabs_html += '\n'
div_inner = fig.to_html(full_html=False, include_plotlyjs="cdn", div_id="plot_" + sym)
divs_html += '