""" Fossati AI Bot - Dashboard Server Execute: python dashboard_server.py Acesse: http://localhost:5000 """ import json import logging import threading from pathlib import Path from datetime import datetime, date from flask import Flask, jsonify, send_file, request from flask_cors import CORS try: from binance.client import Client from binance.exceptions import BinanceAPIException import config binance = Client(config.BINANCE_API_KEY, config.BINANCE_API_SECRET) SYMBOLS = config.SYMBOLS INITIAL_BALANCE = config.INITIAL_BALANCE DAILY_TARGET_PCT = config.DAILY_PROFIT_TARGET SCAN_INTERVAL = config.SCAN_INTERVAL_SECONDS CONSERVATIVE = config.CONSERVATIVE_MODE TRADE_LOG = config.TRADE_LOG_FILE except Exception: binance = None SYMBOLS = [] INITIAL_BALANCE = 1000.0 DAILY_TARGET_PCT = 0.5 SCAN_INTERVAL = 60 CONSERVATIVE = False TRADE_LOG = "trades.json" log = logging.getLogger("fossati.dashboard") app = Flask(__name__) CORS(app) def _load_json(path): p = Path(path) return json.loads(p.read_text()) if p.exists() else {} def _load_trades(): p = Path(TRADE_LOG) if not p.exists(): return [] try: d = json.loads(p.read_text()) return d if isinstance(d, list) else [] except Exception: return [] @app.route("/") def index(): return send_file("dashboard.html") @app.route("/api/stats") def api_stats(): stats = _load_json("daily_stats.json") if not stats: stats = { "date": str(date.today()), "starting_balance": INITIAL_BALANCE, "current_balance": INITIAL_BALANCE, "daily_target": INITIAL_BALANCE * DAILY_TARGET_PCT, "daily_target_pct": DAILY_TARGET_PCT, "realized_pnl": 0, "unrealized_pnl": 0, "total_trades": 0, "winning_trades": 0, "losing_trades": 0, "max_drawdown_today": 0, "target_reached": False, "trading_halted": False, "halt_reason": "", } start = stats.get("starting_balance", 1) curr = stats.get("current_balance", start) target = stats.get("daily_target", start * 0.5) pnl = stats.get("realized_pnl", 0) total = stats.get("total_trades", 0) wins = stats.get("winning_trades", 0) stats["profit_pct"] = round((curr - start) / start * 100, 2) if start else 0 stats["target_progress"] = round(min(pnl / target * 100, 100), 1) if target else 0 stats["win_rate"] = round(wins / total * 100, 1) if total else 0 stats["timestamp"] = datetime.now().isoformat() return jsonify(stats) @app.route("/api/balance") def api_balance(): if binance: try: info = binance.futures_account() for a in info["assets"]: if a["asset"] == "USDT": return jsonify({ "balance": float(a["walletBalance"]), "available": float(a["availableBalance"]), "unrealized": float(a["unrealizedProfit"]), "source": "live", "timestamp": datetime.now().isoformat(), }) except Exception as e: log.warning(f"Binance balance error: {e}") stats = _load_json("daily_stats.json") return jsonify({ "balance": stats.get("current_balance", INITIAL_BALANCE), "available": stats.get("current_balance", INITIAL_BALANCE), "unrealized": 0, "source": "file", "timestamp": datetime.now().isoformat(), }) @app.route("/api/positions") def api_positions(): if binance: try: positions = binance.futures_position_information() result = [] for p in positions: amt = float(p["positionAmt"]) if amt == 0: continue entry = float(p["entryPrice"]) mark = float(p["markPrice"]) pnl = float(p["unRealizedProfit"]) lev = int(p.get("leverage", 1)) margin = abs(amt) * entry / lev if lev else 1 pnl_pct = pnl / margin * 100 if margin else 0 result.append({ "symbol": p["symbol"], "direction": "LONG" if amt > 0 else "SHORT", "size": abs(amt), "entry": entry, "mark_price": mark, "pnl": round(pnl, 2), "pnl_pct": round(pnl_pct, 2), "leverage": lev, "notional": round(abs(amt) * mark, 2), }) return jsonify({"positions": result, "count": len(result)}) except Exception as e: log.warning(f"Binance positions error: {e}") return jsonify({"positions": [], "count": 0}) @app.route("/api/trades") def api_trades(): trades = list(reversed(_load_trades()[-50:])) source = "log" # Fallback: se não há trades fechados pelo bot, mostra execuções reais da Binance if not trades and binance: try: raw = binance.futures_account_trades(limit=50) raw.sort(key=lambda r: r.get("time", 0), reverse=True) for r in raw[:50]: ts = datetime.fromtimestamp(r["time"] / 1000) trades.append({ "symbol": r["symbol"], "direction": "LONG" if r["side"] == "BUY" else "SHORT", "side": r["side"], "entry": float(r["price"]), "exit": float(r["price"]), "size": float(r["qty"]), "notional": round(float(r["quoteQty"]), 2), "pnl": round(float(r.get("realizedPnl", 0)), 4), "fee": round(float(r.get("commission", 0)), 4), "date": ts.date().isoformat(), "closed_at": ts.strftime("%H:%M:%S"), "result": "WIN" if float(r.get("realizedPnl", 0)) > 0 else ("LOSS" if float(r.get("realizedPnl", 0)) < 0 else "—"), }) source = "binance" except Exception as e: log.warning(f"Binance trades error: {e}") return jsonify({"trades": trades, "count": len(trades), "source": source}) @app.route("/api/orders") def api_orders(): """Ordens em aberto na Binance Futures (limit, SL, TP, etc.).""" if not binance: return jsonify({"orders": [], "count": 0, "error": "no binance client"}) try: raw = binance.futures_get_open_orders() orders = [] for o in raw: orders.append({ "symbol": o["symbol"], "side": o["side"], "direction": "LONG" if o["side"] == "BUY" else "SHORT", "type": o["type"], "price": float(o["price"]) if float(o["price"]) > 0 else None, "stop_price": float(o["stopPrice"]) if float(o["stopPrice"]) > 0 else None, "qty": float(o["origQty"]), "executed": float(o["executedQty"]), "reduce_only": o.get("reduceOnly", False), "status": o["status"], "client_id": o.get("clientOrderId", ""), "time": datetime.fromtimestamp(o["time"] / 1000).strftime("%H:%M:%S"), }) orders.sort(key=lambda x: (x["symbol"], x["type"])) return jsonify({"orders": orders, "count": len(orders)}) except Exception as e: log.warning(f"Binance open orders error: {e}") return jsonify({"orders": [], "count": 0, "error": str(e)}) @app.route("/api/equity") def api_equity(): trades = _load_trades() stats = _load_json("daily_stats.json") start = stats.get("starting_balance", INITIAL_BALANCE) today = str(date.today()) points = [{"time": "00:00", "balance": start}] running = start for t in trades: if t.get("date") == today: running += t.get("pnl", 0) points.append({"time": t.get("closed_at", "")[:5], "balance": round(running, 2)}) return jsonify({"equity": points, "start": start}) @app.route("/api/symbols") def api_symbols(): trades = _load_trades() syms = {} for t in trades: s = t.get("symbol", "") if not s: continue if s not in syms: syms[s] = {"symbol": s, "trades": 0, "wins": 0, "pnl": 0.0} syms[s]["trades"] += 1 pnl = t.get("pnl", 0) syms[s]["pnl"] += pnl if pnl > 0: syms[s]["wins"] += 1 for s in syms.values(): t = s["trades"] s["win_rate"] = round(s["wins"] / t * 100, 1) if t else 0 s["pnl"] = round(s["pnl"], 2) return jsonify({"symbols": sorted(syms.values(), key=lambda x: x["pnl"], reverse=True)}) @app.route("/api/status") def api_status(): stats = _load_json("daily_stats.json") if stats.get("target_reached"): status, label = "target_reached", "Meta Atingida" elif stats.get("trading_halted"): status, label = "halted", "Pausado" else: status, label = "running", "Ativo" return jsonify({ "status": status, "label": label, "halt_reason": stats.get("halt_reason", ""), "symbols": len(SYMBOLS), "scan_interval": SCAN_INTERVAL, "conservative": CONSERVATIVE, "timestamp": datetime.now().isoformat(), }) @app.route("/api/scan") def api_scan(): return jsonify(_load_json("scan_state.json") or {"cycle": 0, "symbols": {}}) # ─── BACKTEST ─────────────────────────────────────────────── BACKTEST_RESULT_FILE = "backtest_results.json" _backtest_lock = threading.Lock() _backtest_state = {"running": False, "started_at": None, "error": None} @app.route("/backtest") def backtest_page(): return send_file("backtest.html") @app.route("/api/backtest/last") def api_backtest_last(): p = Path(BACKTEST_RESULT_FILE) if not p.exists(): return jsonify({}), 200 try: return jsonify(json.loads(p.read_text(encoding="utf-8"))) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/backtest/state") def api_backtest_state(): return jsonify(_backtest_state) @app.route("/api/backtest/run", methods=["POST"]) def api_backtest_run(): """Executa o backtest sincronamente e retorna o JSON do resultado. Use rota separada `/api/backtest/state` para polling se quiser tornar async.""" if _backtest_state["running"]: return jsonify({"error": "Já existe um backtest em execução"}), 409 payload = request.get_json(silent=True) or {} symbols = payload.get("symbols") or None if isinstance(symbols, list) and not symbols: symbols = None days = int(payload.get("days") or 90) interval = str(payload.get("interval") or "15m") initial_balance = payload.get("initial_balance") if not _backtest_lock.acquire(blocking=False): return jsonify({"error": "Backtest em execução, aguarde"}), 409 try: _backtest_state.update({"running": True, "started_at": datetime.now().isoformat(), "error": None}) try: from backtest import Backtester except Exception as e: return jsonify({"error": f"Falha ao importar backtest.py: {e}"}), 500 try: bt = Backtester( symbols=symbols, days=days, interval=interval, initial_balance=initial_balance if initial_balance is not None else None, ) bt.run() bt.save(BACKTEST_RESULT_FILE) data = json.loads(Path(BACKTEST_RESULT_FILE).read_text(encoding="utf-8")) return jsonify(data) except Exception as e: log.exception("Backtest falhou") _backtest_state["error"] = str(e) return jsonify({"error": str(e)}), 500 finally: _backtest_state["running"] = False _backtest_lock.release() if __name__ == "__main__": print("\n Fossati AI Bot — Dashboard Server") print(" Acesse: http://localhost:5000\n") app.run(host="0.0.0.0", port=5000, debug=False)