221 lines
9.4 KiB
Python
221 lines
9.4 KiB
Python
"""
|
|
BrainSteel Fin — Audit Agent (v2)
|
|
5-Pillar Compliance Auditor
|
|
Reviews pipeline against all 5 analysis pillars.
|
|
"""
|
|
import os, json, requests
|
|
from datetime import datetime
|
|
|
|
class AuditAgent:
|
|
def __init__(self):
|
|
self.name = "Audit"
|
|
|
|
def send_notification(self, audit_result):
|
|
"""Envia notificação ativa para Telegram/Discord ou grava no log de governança."""
|
|
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
action = audit_result.get("decio_decision", "HOLD")
|
|
score = audit_result.get("compliance_score", 0)
|
|
status = audit_result.get("status", "unknown").upper()
|
|
pnl_pct = audit_result.get("portfolio_pnl_pct", 0)
|
|
bal = audit_result.get("portfolio_balance", 0)
|
|
|
|
msg = (
|
|
f"🔔 [BrainSteel Fin Governança] — {now}\n"
|
|
f"Decisão do Ciclo: {action}\n"
|
|
f"Veredito do Audit: {status} (Score: {score}%)\n"
|
|
f"Saldo Virtual: ${bal:,.2f} | PnL Total: {pnl_pct:+.1f}%\n"
|
|
f"Notas de Qualidade:\n" + "\n".join([f"• {n}" for n in audit_result.get("quality_notes", [])])
|
|
)
|
|
|
|
# 1. Envia Telegram se configurado
|
|
tg_token = os.getenv("TELEGRAM_BOT_TOKEN", "")
|
|
tg_chat = os.getenv("TELEGRAM_CHAT_ID", "")
|
|
if tg_token and tg_chat:
|
|
try:
|
|
requests.post(
|
|
f"https://api.telegram.org/bot{tg_token}/sendMessage",
|
|
json={"chat_id": tg_chat, "text": msg},
|
|
timeout=10
|
|
)
|
|
except: pass
|
|
|
|
# 2. Envia Discord Webhook se configurado
|
|
discord_url = os.getenv("DISCORD_WEBHOOK_URL", "")
|
|
if discord_url:
|
|
try:
|
|
requests.post(discord_url, json={"content": msg}, timeout=10)
|
|
except: pass
|
|
|
|
# 3. Sempre grava no log de notificações de governança
|
|
for path in ["/app/logs/notifications.log", "/data/notifications.log"]:
|
|
try:
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "a", encoding="utf-8") as f:
|
|
f.write(msg + "\n" + "-"*40 + "\n")
|
|
except: pass
|
|
|
|
return msg
|
|
|
|
def audit_pipeline(self, brief_data, decio_data, papert_data):
|
|
checks = []
|
|
|
|
# Check 1: Brief delivered all 3 required points
|
|
briefing = brief_data.get("briefing_for_decio", "")
|
|
has_price = "BTC $" in briefing or "$" in briefing
|
|
has_change = any(x in briefing for x in ["4h:", "24h:", "+", "-"])
|
|
has_sr = "Sup:" in briefing and "Res:" in briefing
|
|
checks.append({
|
|
"check": "Brief 3 pontos",
|
|
"passed": has_price and has_change and has_sr,
|
|
"detail": f"price={has_price} change={has_change} sr={has_sr}"
|
|
})
|
|
|
|
# Check 2: Brief signal was computed correctly
|
|
brief_conf = brief_data.get("confidence", 0)
|
|
brief_signal = brief_data.get("signal", "NEUTRO")
|
|
has_rsi = brief_data.get("rsi") is not None
|
|
has_fg = brief_data.get("fear_greed") is not None
|
|
checks.append({
|
|
"check": "Brief pillars completos",
|
|
"passed": has_rsi and has_fg,
|
|
"detail": f"RSI={brief_data.get('rsi')} F&G={brief_data.get('fear_greed')} news={brief_data.get('news_sentiment','?')}"
|
|
})
|
|
|
|
# Check 3: Decio respected 75% rule
|
|
decio_action = decio_data.get("action", "HOLD")
|
|
decio_conf = decio_data.get("confidence", 0)
|
|
sl = decio_data.get("stop_loss", 0)
|
|
tp = decio_data.get("take_profit", 0)
|
|
|
|
respected_75 = True
|
|
if decio_conf < 75 and decio_action != "HOLD":
|
|
respected_75 = False
|
|
checks.append({
|
|
"check": "Decio regra 75%",
|
|
"passed": respected_75,
|
|
"detail": f"conf={decio_conf}% action={decio_action}"
|
|
})
|
|
|
|
# Check 4: Decio SL/TP correct if BUY/SELL
|
|
price = brief_data.get("price", 0)
|
|
if decio_action in ("BUY", "SELL"):
|
|
sl_correct = (decio_action == "BUY" and abs(sl - price * 0.98) < price * 0.01) or \
|
|
(decio_action == "SELL" and abs(sl - price * 1.02) < price * 0.01)
|
|
tp_correct = (decio_action == "BUY" and abs(tp - price * 1.05) < price * 0.01) or \
|
|
(decio_action == "SELL" and abs(tp - price * 0.95) < price * 0.01)
|
|
else:
|
|
sl_correct = (sl == 0 and tp == 0)
|
|
checks.append({
|
|
"check": "Decio SL/TP 2%/5%",
|
|
"passed": sl_correct,
|
|
"detail": f"SL=${sl} TP=${tp} price=${price}"
|
|
})
|
|
|
|
# Check 5: PaperT logged correctly
|
|
papert_logged = papert_data.get("log_appended", False)
|
|
papert_status = papert_data.get("status", "")
|
|
checks.append({
|
|
"check": "PaperT executou log",
|
|
"passed": papert_logged and papert_status == "validated",
|
|
"detail": f"logged={papert_logged} status={papert_status}"
|
|
})
|
|
|
|
# Check 6: Portfolio updated if BUY/SELL
|
|
portfolio = papert_data.get("portfolio", {})
|
|
if decio_action in ("BUY", "SELL"):
|
|
pf_updated = portfolio.get("total_trades", 0) > 0
|
|
else:
|
|
pf_updated = True # No trade = no portfolio change needed
|
|
checks.append({
|
|
"check": "Portfolio atualizado",
|
|
"passed": pf_updated,
|
|
"detail": f"trades={portfolio.get('total_trades',0)} balance=${portfolio.get('current_balance','?')}"
|
|
})
|
|
|
|
# Score
|
|
passed_count = sum(1 for c in checks if c["passed"])
|
|
total = len(checks)
|
|
score = round(passed_count / total * 100, 1)
|
|
|
|
# Signal quality assessment
|
|
net = brief_data.get("net_signal", 0)
|
|
bullish = brief_data.get("bullish_signals", 0)
|
|
bearish = brief_data.get("bearish_signals", 0)
|
|
rsi = brief_data.get("rsi", 50)
|
|
|
|
quality_notes = []
|
|
if rsi > 70: quality_notes.append("RSI overbought — cautela")
|
|
if rsi < 30: quality_notes.append("RSI oversold — oportunidade")
|
|
if net > 2: quality_notes.append(f"Sinal bullish forte (net={net})")
|
|
if net < -2: quality_notes.append(f"Sinal bearish forte (net={net})")
|
|
if decio_action == "HOLD" and brief_data.get("confidence", 0) < 75:
|
|
quality_notes.append("HOLD correto — confiança baixa preservou capital")
|
|
|
|
res = {
|
|
"audit_id": f"AUD-{datetime.now().strftime('%Y%m%d%H%M%S')}",
|
|
"timestamp": datetime.now().isoformat(),
|
|
"checks": checks,
|
|
"compliance_score": score,
|
|
"status": "approved" if score >= 83 else "flagged" if score >= 66 else "rejected",
|
|
"passed_count": passed_count,
|
|
"total_checks": total,
|
|
"quality_notes": quality_notes,
|
|
"brief_summary": brief_data.get("summary", "")[:100],
|
|
"decio_decision": decio_action,
|
|
"decio_confidence": decio_conf,
|
|
"portfolio_balance": portfolio.get("current_balance") or portfolio.get("balance", 0),
|
|
"portfolio_pnl": portfolio.get("total_pnl", 0),
|
|
"portfolio_pnl_pct": portfolio.get("total_pnl_pct", 0),
|
|
}
|
|
self.send_notification(res)
|
|
return res
|
|
|
|
def generate_llm_report(self, brief_data, decio_data):
|
|
if not os.getenv("PAPERT_API_KEY") and not os.getenv("OPENROUTER_API_KEY", ""):
|
|
return {"report": "OpenRouter key not available", "status": "local_only"}
|
|
try:
|
|
prompt = f"""Você é o Audit, auditor de compliance da BrainSteel Fin.
|
|
Gere um relatório de auditoria em português，分析昨日操作质量:
|
|
|
|
Brief: {brief_data.get('summary','N/A')}
|
|
Signal: {brief_data.get('signal','?')} ({brief_data.get('confidence','?')}% conf)
|
|
RSI: {brief_data.get('rsi','?')} | F&G: {brief_data.get('fear_greed_class','?')}
|
|
Bullish signals: {brief_data.get('bullish_signals',0)} | Bearish: {brief_data.get('bearish_signals',0)}
|
|
|
|
Decio: {decio_data.get('action','?')} conf {decio_data.get('confidence','?')}%
|
|
Justification: {decio_data.get('justification','?')[:100]}
|
|
|
|
Responda em JSON:
|
|
{{"verdict": "APROVADO|REPROVADO|FLAG",
|
|
"risks": ["risk1","risk2"],
|
|
"recommendations": ["rec1","rec2"],
|
|
"score": 0-100,
|
|
"summary_pt": "resumo em português"}}
|
|
"""
|
|
headers = {
|
|
"Authorization": "Bearer " + (os.getenv("PAPERT_API_KEY") or os.getenv("OPENROUTER_API_KEY", "")),
|
|
"Content-Type": "application/json",
|
|
"HTTP-Referer": "https://brainsteel.fin",
|
|
"X-Title": "BrainSteel Fin"
|
|
}
|
|
r = requests.post(
|
|
"https://openrouter.ai/api/v1/chat/completions",
|
|
json={"model": "deepseek/deepseek-v4-flash:free", "messages": [{"role": "user", "content": prompt}], "max_tokens": 400},
|
|
headers=headers, timeout=30
|
|
)
|
|
if r.status_code == 200:
|
|
resp_data = r.json()
|
|
content = resp_data["choices"][0]["message"]["content"]
|
|
usage = resp_data.get("usage", {})
|
|
if usage:
|
|
try:
|
|
from agents.token_tracker import log_tokens
|
|
log_tokens("Audit", "deepseek/deepseek-v4-flash:free", usage)
|
|
except:
|
|
pass
|
|
content = content.strip("` \n")
|
|
if content.startswith("json"): content = content[4:]
|
|
return {"report": content, "status": "generated"}
|
|
except:
|
|
pass
|
|
return {"report": "Audit local", "status": "error"} |