"""
mini_proxy.py - Reverse proxy Tailscale Funnel (porta 10000)
Mavis 2026-07-21 - v4: layout pulito + 2 PULSANTI EMERGENZA fissi in alto a destra.
Route:
- / → landing HTML (PC LOCALE / VPS / SERVIZI ESTERNI)
- /webhook → http://127.0.0.1:5580 (webhook_receiver.py)
- /stats → http://127.0.0.1:5503 (stats_dashboard.py)
- /square → http://127.0.0.1:5504 (square_monitor.py)
- /tableau → file statico G:\\AI TRADING ENGINE\\live_deploy\\tableau_de_bord.html
- /healthz → 200 OK con info processi
- /status → JSON dettagliato processi
- /emergency/stop?token=X → KILL processi trading
- /emergency/restart?token=X → STOP + START processi trading
EMERGENZA (Mattia 21/07 15:36): "in live una situazione del genre mi brucio tutto il capitale".
I 2 pulsanti sono STICKY in alto a destra (sempre visibili anche scrollando).
Tailscale Funnel: https://mabest.tail2b1710.ts.net → http://127.0.0.1:10000
LINK UNICO PER MATTIA: https://mabest.tail2b1710.ts.net
"""
import os
import sys
import time
import json
import logging
import subprocess
import socket
import requests
from pathlib import Path
from flask import Flask, Response, redirect, request
from datetime import datetime
LIVE_DEPLOY = Path(r'G:\AI TRADING ENGINE\live_deploy')
LOG_DIR = LIVE_DEPLOY / 'logs'
LOG_FILE = LOG_DIR / 'mini_proxy.log'
PID_FILE = Path(r'C:\Users\Mattia\AppData\Local\Temp\mini_proxy.pid')
EMERGENCY_TOKEN = "MATTIA_STOP_2026"
TRADING_SERVICES = [
"mavis_supervisor_3.py",
"webhook_receiver.py",
"rettangolo_runner.py",
"trailing_stop_watchdog.py",
"generate_tableau.py",
]
PYTHON_EXE = r"C:\Users\Mattia\AppData\Local\Programs\Python\Python314\python.exe"
LOG_DIR.mkdir(parents=True, exist_ok=True)
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
)
log = logging.getLogger()
app = Flask(__name__)
WEBHOOK_PORT = 5580
STATS_PORT = 5503
NEWDASH_PORT = 5510
SQUARE_PORT = 5504
LANDING_HTML = """
Mavis Trading Engine - Hub
🤖 Mavis Trading Engine
Hub centrale — aggiornato {timestamp} · refresh automatico 60s
Link principale: https://mabest.tail2b1710.ts.net
📍 PC LOCALE (mabest) — strategie live
☁️ VPS HETZNER (charter-live-vps) — 2° account, monitor mode
🌐 SERVIZI ESTERNI
"""
def _proxy(target_url: str, timeout: int = 30) -> Response:
try:
if request.query_string:
sep = '&' if '?' in target_url else '?'
target_url = f"{target_url}{sep}{request.query_string.decode('utf-8')}"
resp = requests.request(
method=request.method,
url=target_url,
headers={k: v for k, v in request.headers if k.lower() not in ('host', 'content-length')},
data=request.get_data(),
cookies=request.cookies,
allow_redirects=False,
timeout=timeout,
)
excluded_headers = ['content-encoding', 'content-length', 'transfer-encoding', 'connection']
headers = [(k, v) for k, v in resp.headers.items() if k.lower() not in excluded_headers]
body = resp.content
# Inject MTE status banner in HTML responses
ctype = resp.headers.get('content-type', '').lower()
if 'text/html' in ctype and b'', banner.encode('utf-8') + b' ', 1)
except Exception as e:
log.warning(f"banner injection failed: {e}")
return Response(body, resp.status_code, headers)
except requests.exceptions.RequestException as e:
log.error(f"proxy error to {target_url}: {e}")
return Response(f"Backend error: {e}", 502)
def _check_port(port: int, timeout: float = 1.5) -> bool:
"""True se la porta locale è in ascolto."""
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(timeout)
s.connect(("127.0.0.1", port))
return True
except Exception:
return False
def _check_service(name: str, port: int, proc_name: str = None) -> bool:
"""Verifica servizio: combinazione processo Python + porta in LISTEN.
Se proc_name specificato, richiede ENTRAMBI (esclude falsi positivi come tailscaled sulla 10000).
"""
port_ok = _check_port(port)
if not proc_name:
return port_ok
try:
procs = _list_python_processes()
proc_ok = any(p["name"] == proc_name for p in procs)
except Exception:
proc_ok = False
return port_ok and proc_ok
def _check_mte_services() -> dict:
"""Stato aggregato servizi Mavis Trading Engine. Ritorna dict con stato per servizio."""
services = {
"webhook": {"port": WEBHOOK_PORT, "label": "Webhook 5580", "proc": "webhook_receiver.py"},
"v3_dashboard": {"port": NEWDASH_PORT, "label": "V3 Dashboard 5510", "proc": "stats_dashboard_live_v3.py"},
"old_dashboard": {"port": 5503, "label": "Old Dashboard 5503", "proc": "stats_dashboard.py"},
"square": {"port": 5504, "label": "Square Monitor 5504", "proc": "square_monitor.py"},
"mini_proxy": {"port": 10000, "label": "mini_proxy 10000", "proc": "mini_proxy.py"},
}
result = {"services": {}, "ok_count": 0, "total": len(services), "status": "ok"}
for name, info in services.items():
up = _check_service(name, info["port"], info.get("proc"))
info["up"] = up
info["status"] = "UP" if up else "DOWN"
result["services"][name] = info
if up:
result["ok_count"] += 1
if result["ok_count"] == 0:
result["status"] = "red"
elif result["ok_count"] < result["total"]:
result["status"] = "yellow"
else:
result["status"] = "green"
return result
def _inject_banner(html: str) -> str:
"""Inietta banner semaforo MTE prima di . Idempotente (no doppia injection)."""
if 'id="mte-semaforo"' in html:
return html
banner = _render_mte_banner()
lower = html.lower()
if '' in lower:
idx = lower.rfind('')
return html[:idx] + banner + html[idx:]
return html + banner
def _render_mte_banner() -> str:
"""Banner HTML + JS per semaforo stato MTE. Auto-refresh ogni 30s."""
return (
'