""" BrainSteel Fin — Flask Dashboard Virtual Agent Office with anime-style agents 4-Agent BTC Trading Pipeline """ import os, json, sqlite3, threading, hashlib, sys from pathlib import Path _env_p = Path(__file__).parent / "data" / ".env" if _env_p.exists(): for line in _env_p.read_text().splitlines(): line = line.strip() if line and "=" in line and not line.startswith("#"): k, v = line.split("=", 1) os.environ[k.strip()] = v.strip() from datetime import datetime, timedelta, date from flask import Flask, render_template, jsonify, request, Response from agents.brief import BriefAgent from agents.decio import DecioAgent from agents.papert import PapertAgent from agents.audit import AuditAgent app = Flask(__name__) app.secret_key = os.getenv("SECRET_KEY", "brainsteel-fin-secret-2025") # Paths BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(BASE_DIR, "data") LOG_DIR = os.path.join(BASE_DIR, "logs") os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(LOG_DIR, exist_ok=True) DB_PATH = os.path.join(DATA_DIR, "executions.db") AGENTS_DB = os.path.join(DATA_DIR, "agents.db") # ── Databases ───────────────────────────────────────────────────────────────── def init_db(): conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute(""" CREATE TABLE IF NOT EXISTS executions ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, agent TEXT, action TEXT, result TEXT, status TEXT ) """) c.execute(""" CREATE TABLE IF NOT EXISTS pipeline_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, started_at TEXT, finished_at TEXT, brief_result TEXT, decio_decision TEXT, papert_result TEXT, audit_report TEXT, status TEXT ) """) c.execute(""" CREATE TABLE IF NOT EXISTS token_usage ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp TEXT, agent TEXT, model TEXT, prompt_tokens INTEGER, completion_tokens INTEGER, total_tokens INTEGER, cost_usd REAL ) """) conn.commit() conn.close() def init_agents_db(): """Agent registry — onboarding + state.""" conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() c.execute(""" CREATE TABLE IF NOT EXISTS agents ( id TEXT PRIMARY KEY, name TEXT, role TEXT, intent TEXT, model TEXT, provider TEXT, status TEXT DEFAULT 'idle', last_run TEXT, cycle_count INTEGER DEFAULT 0, success_count INTEGER DEFAULT 0, fail_count INTEGER DEFAULT 0, avg_duration_ms INTEGER DEFAULT 0, created_at TEXT, updated_at TEXT ) """) c.execute(""" CREATE TABLE IF NOT EXISTS agent_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, agent_id TEXT, timestamp TEXT, event TEXT, duration_ms INTEGER, result TEXT, FOREIGN KEY (agent_id) REFERENCES agents(id) ) """) c.execute(""" CREATE TABLE IF NOT EXISTS services ( id TEXT PRIMARY KEY, name TEXT, type TEXT, endpoint TEXT, status TEXT DEFAULT 'unknown', last_check TEXT, metadata TEXT ) """) conn.commit() conn.close() def seed_agents(): """Seed default agents on first run.""" conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() # Check if already seeded row = c.execute("SELECT COUNT(*) FROM agents").fetchone()[0] if row == 0: defaults = [ ("brief", "Brief", "Market Intelligence Analyst", "Analisa dados de mercado BTC/CRIPTOS. Coleta preços, notícias, indicadores técnicos. Fornece resumo resumido ao Decio.", "google/gemma-4-31b-it:free", "openrouter", "idle"), ("decio", "Decio", "Estrategista Soberano (CEO)", "CEO Estrategista. Recebe briefing do Brief e decide BUY/SELL/HOLD para BTC. Define stop_loss e take_profit.", "deepseek/deepseek-v4-flash:free", "openrouter", "idle"), ("papert", "PaperT", "Executor de Baixa Latência", "Executor. Valida decisão do Decio e executa ordens no Binance (simulação). Gera log de execução.", "google/gemma-4-26b-a4b-it:free", "openrouter", "idle"), ("audit", "Audit", "Auditor & Compliance", "Auditor. Revisa cada ciclo pipeline. Valida decisões, verifica compliance, gera relatório de risco.", "deepseek/deepseek-v4-flash:free", "openrouter", "idle"), ] now = datetime.now().isoformat() for a in defaults: c.execute("INSERT INTO agents (id,name,role,intent,model,provider,status,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?)", (*a, now, now)) conn.commit() conn.close() init_db() init_agents_db() seed_agents() # ── Agent State ──────────────────────────────────────────────────────────────── class AgentState: def __init__(self): self.lock = threading.Lock() self.brief = {"status": "idle", "message": "Aguardando...", "last_run": None} self.decio = {"status": "idle", "message": "Aguardando...", "last_run": None} self.papert = {"status": "idle", "message": "Aguardando...", "last_run": None} self.audit = {"status": "idle", "message": "Aguardando...", "last_run": None} self.pipeline_running = False self.last_pipeline = None def _ensure_agent(self, agent): """Ensure instance attribute exists (getattr returns CLASS attr otherwise).""" if not hasattr(self, agent) or not isinstance(getattr(self, agent), dict): setattr(self, agent, {"status": "idle", "message": "Aguardando...", "last_run": None}) def update(self, agent, status, message): with self.lock: self._ensure_agent(agent) d = getattr(self, agent) d["status"] = status d["message"] = message d["last_run"] = datetime.now().isoformat() def set_pipeline_running(self, val): with self.lock: self.pipeline_running = val if val: self.last_pipeline = datetime.now().isoformat() def get_all(self): with self.lock: return { "brief": self.brief.copy(), "decio": self.decio.copy(), "papert": self.papert.copy(), "audit": self.audit.copy(), "pipeline_running": self.pipeline_running, "last_pipeline": self.last_pipeline } state = AgentState() # ── Logging ─────────────────────────────────────────────────────────────────── def log_execution(agent, action, result, status="success"): conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute("INSERT INTO executions (timestamp, agent, action, result, status) VALUES (?, ?, ?, ?, ?)", (datetime.now().isoformat(), agent, action, result, status)) conn.commit() conn.close() # Also log to agent_logs _log_agent_event(agent, "execution", result) def _log_agent_event(agent_id, event, result, duration_ms=0): conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() c.execute("INSERT INTO agent_logs (agent_id, timestamp, event, duration_ms, result) VALUES (?, ?, ?, ?, ?)", (agent_id, datetime.now().isoformat(), event, duration_ms, result)) conn.commit() conn.close() # ── Pipeline Execution ──────────────────────────────────────────────────────── def run_pipeline(): state.set_pipeline_running(True) state.update("brief", "working", "Analisando mercado...") state.update("decio", "waiting", "Aguardando análise...") state.update("papert", "waiting", "Aguardando decisão...") state.update("audit", "waiting", "Aguardando ciclo...") start = datetime.now() brief_data, decio_data, papert_data, audit_data = {}, {}, {}, {} try: # ── Step 1: Brief ── t0 = datetime.now() state.update("brief", "working", "Buscando dados da Binance...") brief = BriefAgent() brief_data = brief.run() t1 = datetime.now() state.update("brief", "done", brief_data.get("summary", "")[:80]) # Also persist brief key metrics to state for /api/state brief_summary = brief_data.get("summary", "") state.brief.update({ "price": brief_data.get("price"), "signal": brief_data.get("signal"), "confidence": brief_data.get("confidence"), "rsi": brief_data.get("rsi"), "fear_greed": brief_data.get("fear_greed"), "fear_greed_class": brief_data.get("fear_greed_class"), "change_4h": brief_data.get("change_4h"), "change_24h": brief_data.get("change_24h"), "funding_rate": brief_data.get("funding_rate"), "long_ratio": brief_data.get("long_ratio"), "support": brief_data.get("support"), "resistance": brief_data.get("resistance"), "bullish_signals": brief_data.get("bullish_signals"), "bearish_signals": brief_data.get("bearish_signals"), "net_signal": brief_data.get("net_signal"), "news_sentiment": brief_data.get("news_sentiment"), "macd_histogram": brief_data.get("macd_histogram"), "btc_dominance": brief_data.get("btc_dominance"), }) log_execution("brief", "market_analysis", brief_summary, "success") _update_agent_stats("brief", t1 - t0, True) # ── Step 2: Decio ── t0 = datetime.now() state.update("decio", "working", "Formando estratégia...") decio = DecioAgent(brief_data) decio_data = decio.run() t1 = datetime.now() state.update("decio", "done", decio_data.get("decision", "HOLD")) state.decio.update({ "action": decio_data.get("action"), "confidence": decio_data.get("confidence"), "stop_loss": decio_data.get("stop_loss"), "take_profit": decio_data.get("take_profit"), "justification": decio_data.get("justification", "")[:100], }) log_execution("decio", "strategy_decision", decio_data.get("decision", "HOLD"), "success") _update_agent_stats("decio", t1 - t0, True) # ── Step 3: PaperT ── t0 = datetime.now() state.update("papert", "working", "Executando ordens...") papert = PapertAgent(decio_data) papert_data = papert.run() t1 = datetime.now() state.update("papert", "done", papert_data.get("result", "")[:80]) log_execution("papert", "order_execution", papert_data.get("result", ""), "success") _update_agent_stats("papert", t1 - t0, True) # ── Step 4: Audit ── t0 = datetime.now() state.update("audit", "working", "Auditando compliance...") audit = AuditAgent() audit_data = audit.audit_pipeline(brief_data, decio_data, papert_data) t1 = datetime.now() state.update("audit", "done", audit_data.get("summary", "")[:80]) log_execution("audit", "compliance_check", audit_data.get("summary", ""), "success") _update_agent_stats("audit", t1 - t0, True) # ── Save pipeline run ── conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute(""" INSERT INTO pipeline_runs (started_at, finished_at, brief_result, decio_decision, papert_result, audit_report, status) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( start.isoformat(), datetime.now().isoformat(), brief_data.get("summary", ""), decio_data.get("decision", "HOLD"), papert_data.get("result", ""), json.dumps(audit_data), "success" )) conn.commit() conn.close() except Exception as e: log_execution("pipeline", "error", str(e), "error") state.update("brief", "error", f"Erro: {str(e)[:80]}") _update_agent_stats("brief", datetime.now() - start, False) finally: state.set_pipeline_running(False) state.update("decio", "idle", "Aguardando...") state.update("papert", "idle", "Aguardando...") state.update("audit", "idle", "Aguardando...") def _update_agent_stats(agent_id, duration, success): conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() now = datetime.now().isoformat() dur = int(duration.total_seconds() * 1000) c.execute("UPDATE agents SET last_run=?, updated_at=? WHERE id=?", (now, now, agent_id)) c.execute("UPDATE agents SET cycle_count=cycle_count+1 WHERE id=?", (agent_id,)) if success: c.execute("UPDATE agents SET success_count=success_count+1 WHERE id=?", (agent_id,)) else: c.execute("UPDATE agents SET fail_count=fail_count+1 WHERE id=?", (agent_id,)) # Rolling avg duration row = c.execute("SELECT avg_duration_ms, cycle_count FROM agents WHERE id=?", (agent_id,)).fetchone() if row and row[1]: old_avg = row[0] or 0 count = row[1] new_avg = int((old_avg * (count - 1) + dur) / count) c.execute("UPDATE agents SET avg_duration_ms=? WHERE id=?", (new_avg, agent_id)) conn.commit() conn.close() # ── Routes ──────────────────────────────────────────────────────────────────── @app.route("/") def index(): return render_template("index.html") @app.route("/api/state") def api_state(): return jsonify(state.get_all()) @app.route("/api/run", methods=["POST"]) def api_run(): if state.pipeline_running: return jsonify({"error": "Pipeline já em execução"}), 409 thread = threading.Thread(target=run_pipeline) thread.start() return jsonify({"status": "started", "message": "Pipeline iniciado"}) @app.route("/api/history") def api_history(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row c = conn.cursor() rows = c.execute("SELECT * FROM executions ORDER BY timestamp DESC LIMIT 50").fetchall() conn.close() return jsonify([dict(r) for r in rows]) @app.route("/api/pipeline_history") def api_pipeline_history(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row c = conn.cursor() rows = c.execute("SELECT * FROM pipeline_runs ORDER BY started_at DESC LIMIT 50").fetchall() conn.close() return jsonify([dict(r) for r in rows]) # ── Token Tracking ──────────────────────────────────────────────────────────── @app.route("/api/token_stats", methods=["GET"]) def api_token_stats(): """Daily token bar chart data per agent — for dashboard.""" days = int(request.args.get("days", 7)) from agents.token_tracker import get_token_stats return jsonify(get_token_stats(days=days)) @app.route("/api/token_totals", methods=["GET"]) def api_token_totals(): """All-time token totals per agent.""" conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute(""" SELECT agent, SUM(prompt_tokens) as p, SUM(completion_tokens) as c, SUM(total_tokens) as t, SUM(cost_usd) as cost, COUNT(*) as calls FROM token_usage GROUP BY agent """) rows = c.fetchall() conn.close() return jsonify([{"agent": r[0], "prompt": r[1], "completion": r[2], "total": r[3], "cost": r[4], "calls": r[5]} for r in rows]) # ── Item 1: Agent Registry ──────────────────────────────────────────────────── @app.route("/api/agents", methods=["GET"]) def get_agents(): """List all registered agents with full status.""" conn = sqlite3.connect(AGENTS_DB) conn.row_factory = sqlite3.Row c = conn.cursor() rows = c.execute("SELECT * FROM agents ORDER BY id").fetchall() conn.close() return jsonify([dict(r) for r in rows]) @app.route("/api/agents/", methods=["GET"]) def get_agent(agent_id): conn = sqlite3.connect(AGENTS_DB) conn.row_factory = sqlite3.Row c = conn.cursor() row = c.execute("SELECT * FROM agents WHERE id=?", (agent_id,)).fetchone() conn.close() if not row: return jsonify({"error": "Agent not found"}), 404 return jsonify(dict(row)) @app.route("/api/agents/", methods=["PUT"]) def update_agent(agent_id): """Update agent config (model, provider, status, intent).""" data = request.json conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() fields = [] values = [] for field in ("name", "role", "intent", "model", "provider", "status"): if field in data: fields.append(f"{field}=?") values.append(data[field]) if fields: fields.append("updated_at=?") values.append(datetime.now().isoformat()) values.append(agent_id) c.execute(f"UPDATE agents SET {','.join(fields)} WHERE id=?", values) conn.commit() conn.close() return jsonify({"status": "updated", "id": agent_id}) @app.route("/api/agents/", methods=["DELETE"]) def delete_agent(agent_id): """Remove agent from registry.""" conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() c.execute("DELETE FROM agent_logs WHERE agent_id=?", (agent_id,)) c.execute("DELETE FROM agents WHERE id=?", (agent_id,)) conn.commit() conn.close() return jsonify({"status": "deleted", "id": agent_id}) @app.route("/api/agents", methods=["POST"]) def register_agent(): """Onboard a new agent.""" data = request.json if not data.get("id") or not data.get("name"): return jsonify({"error": "id and name required"}), 400 conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() now = datetime.now().isoformat() c.execute(""" INSERT OR REPLACE INTO agents (id, name, role, intent, model, provider, status, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, COALESCE(?, 'idle'), ?, ?) """, ( data["id"], data["name"], data.get("role", ""), data.get("intent", ""), data.get("model", "unknown"), data.get("provider", "unknown"), data.get("status"), now, now )) conn.commit() conn.close() return jsonify({"status": "registered", "id": data["id"]}), 201 # ── Item 2: Service Metrics ─────────────────────────────────────────────────── @app.route("/api/agent_metrics", methods=["GET"]) def get_agent_metrics(): """Dashboard metrics for all agents.""" conn = sqlite3.connect(AGENTS_DB) conn.row_factory = sqlite3.Row c = conn.cursor() rows = c.execute("SELECT * FROM agents ORDER BY id").fetchall() conn.close() # Merge with pipeline stats from execution DB conn2 = sqlite3.connect(DB_PATH) conn2.row_factory = sqlite3.Row c2 = conn2.cursor() result = [] for row in rows: r = dict(row) # Get last execution from executions table last = c2.execute( "SELECT timestamp, action, result FROM executions WHERE agent=? ORDER BY timestamp DESC LIMIT 1", (r["id"],) ).fetchone() if last: r["last_action"] = dict(last) # Success rate total = r.get("cycle_count", 0) or 1 ok = r.get("success_count", 0) or 0 r["success_rate"] = round(ok / total * 100, 1) result.append(r) conn2.close() return jsonify(result) @app.route("/api/agent_logs/", methods=["GET"]) def get_agent_logs(agent_id): """Event log for specific agent.""" conn = sqlite3.connect(AGENTS_DB) conn.row_factory = sqlite3.Row c = conn.cursor() limit = request.args.get("limit", 50, type=int) rows = c.execute( "SELECT * FROM agent_logs WHERE agent_id=? ORDER BY timestamp DESC LIMIT ?", (agent_id, limit) ).fetchall() conn.close() return jsonify([dict(r) for r in rows]) # ── Item 3: Auto-Discovery ───────────────────────────────────────────────────── @app.route("/api/services", methods=["GET"]) def get_services(): """List all registered services.""" conn = sqlite3.connect(AGENTS_DB) conn.row_factory = sqlite3.Row c = conn.cursor() rows = c.execute("SELECT * FROM services ORDER BY name").fetchall() conn.close() return jsonify([dict(r) for r in rows]) @app.route("/api/services", methods=["POST"]) def register_service(): """Auto-discover / register a new service.""" data = request.json if not data.get("name"): return jsonify({"error": "name required"}), 400 service_id = data.get("id") or hashlib.md5(data["name"].encode()).hexdigest()[:12] conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() c.execute(""" INSERT OR REPLACE INTO services (id, name, type, endpoint, status, last_check, metadata) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( service_id, data["name"], data.get("type", "unknown"), data.get("endpoint", ""), data.get("status", "unknown"), datetime.now().isoformat(), json.dumps(data.get("metadata", {})) )) conn.commit() conn.close() return jsonify({"status": "registered", "id": service_id}), 201 @app.route("/api/services//ping", methods=["POST"]) def ping_service(service_id): """Health check a service endpoint.""" import requests as req conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() row = c.execute("SELECT endpoint FROM services WHERE id=?", (service_id,)).fetchone() conn.close() if not row or not row[0]: return jsonify({"error": "No endpoint for service"}), 404 endpoint = row[0] try: r = req.get(endpoint, timeout=5) status = "healthy" if r.status_code < 500 else "degraded" except Exception as e: status = "unreachable" conn = sqlite3.connect(AGENTS_DB) c = conn.cursor() c.execute("UPDATE services SET status=?, last_check=? WHERE id=?", (status, datetime.now().isoformat(), service_id)) conn.commit() conn.close() return jsonify({"id": service_id, "status": status, "endpoint": endpoint}) # ── Item 4: Pipeline Visual ────────────────────────────────────────────────── @app.route("/api/pipeline/visual") def pipeline_visual(): """Real-time pipeline flow with metrics.""" conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row c = conn.cursor() # Last pipeline run last = c.execute("SELECT * FROM pipeline_runs ORDER BY started_at DESC LIMIT 1").fetchone() last_5 = c.execute("SELECT * FROM pipeline_runs ORDER BY started_at DESC LIMIT 5").fetchall() # Execution summary stats stats = c.execute(""" SELECT agent, COUNT(*) as total, SUM(CASE WHEN status='success' THEN 1 ELSE 0 END) as ok, MAX(timestamp) as last_ts FROM executions GROUP BY agent """).fetchall() conn.close() # Build pipeline nodes nodes = [ {"id": "brief", "label": "Brief", "role": "Market Intelligence", "emoji": "📊", "color": "#4fc3f7"}, {"id": "decio", "label": "Decio", "role": "Estrategista (CEO)", "emoji": "🎯", "color": "#ffd54f"}, {"id": "papert", "label": "PaperT", "role": "Executor", "emoji": "⚡", "color": "#69f0ae"}, {"id": "audit", "label": "Audit", "role": "Auditor Compliance", "emoji": "🔍", "color": "#b388ff"}, ] # Current state from state object current = state.get_all() # Enrich nodes with current state and stats for node in nodes: aid = node["id"] node.update(current.get(aid, {"status": "idle", "message": ""})) # Stats from DB for s in stats: if dict(s)["agent"] == aid: node["total_runs"] = dict(s)["total"] node["success_runs"] = dict(s)["ok"] node["last_execution"] = dict(s)["last_ts"] # Last pipeline decision last_decision = None if last: ld = dict(last) last_decision = ld.get("decio_decision", "") last_brief = ld.get("brief_result", "")[:100] return jsonify({ "nodes": nodes, "last_decision": last_decision, "last_brief": last_brief, "last_pipeline_ts": last["started_at"] if last else None, "recent_runs": [dict(r) for r in last_5], "stats": [dict(s) for s in stats] }) # ── Portfolio API ───────────────────────────────────────────────────────────── @app.route("/api/portfolio", methods=["GET"]) def get_portfolio(): """Returns current portfolio stats + chart data.""" sys.path.insert(0, "/app/agents") try: import papert as papert_module pf = papert_module.Portfolio() stats = pf.get_stats() # Recent trades conn = sqlite3.connect(str(papert_module.PORTFOLIO_DB)) conn.row_factory = sqlite3.Row c = conn.cursor() recent_trades = c.execute( "SELECT * FROM trades ORDER BY created_at DESC LIMIT 20" ).fetchall() conn.close() stats["recent_trades"] = [dict(t) for t in recent_trades] stats["initial_balance"] = papert_module.INITIAL_BALANCE return jsonify(stats) except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/portfolio/reset", methods=["POST"]) def reset_portfolio(): """Reset portfolio to initial $10.000.""" sys.path.insert(0, "/app/agents") try: import papert as papert_module pf = papert_module.Portfolio() conn = sqlite3.connect(str(papert_module.PORTFOLIO_DB)) c = conn.cursor() c.execute("DELETE FROM trades") c.execute("DELETE FROM daily_snapshot") today = date.today().isoformat() c.execute("DELETE FROM portfolio WHERE date=?", (today,)) c.execute("INSERT INTO portfolio (date,balance,btc_held,created_at) VALUES (?,?,?,?)", (today, papert_module.INITIAL_BALANCE, 0.0, datetime.now().isoformat())) conn.commit() conn.close() return jsonify({"status": "reset", "balance": papert_module.INITIAL_BALANCE}) except Exception as e: return jsonify({"error": str(e)}), 500 if __name__ == "__main__": app.run(host="0.0.0.0", port=3100, debug=False)