""" BrainSteel Fin — Brief Agent (v2) Market Intelligence Analyst — Super Financial Agent 5 Pillars: On-Chain | Derivatives | Sentiment | Technical | Macro """ import os, json, requests, re from datetime import datetime, timedelta from collections import defaultdict # ── Data Sources ───────────────────────────────────────────────────────────── GLASSNODE_API = os.getenv("GLASSNODE_API_KEY", "") COINGLASS_API = os.getenv("COINGLASS_API_KEY", "") SANTIMENT_API = os.getenv("SANTIMENT_API_KEY", "") class BriefAgent: def __init__(self): self.name = "Brief" self.openrouter_key = os.getenv("BRIEF_API_KEY", os.getenv("OPENROUTER_API_KEY", "")) self.openrouter_url = "https://openrouter.ai/api/v1/chat/completions" self.model = "deepseek/deepseek-v4-flash:free" # ═══════════════════════════════════════════════════════ # PILLAR 1 — ON-CHAIN ANALYSIS # ═══════════════════════════════════════════════════════ def _onchain_bitflyer(self): """On-chain data via Blockchain.com public API + mempool.space.""" result = {} try: # BTC UTXO set — circulating supply estimate r1 = requests.get("https://blockchain.info/q/estimatedbtcsupply", timeout=8) if r1.status_code == 200: result["btc_supply"] = float(r1.text.strip()) except: pass try: # Market cap r2 = requests.get("https://api.blockchain.com/v3/index/c BTC/information", timeout=8) if r2.status_code == 200: data = r2.json() result["market_cap"] = data.get("market_cap") except: pass try: # Mempool congestion — fee estimation r3 = requests.get("https://mempool.space/api/v1/fees/recommended", timeout=8) if r3.status_code == 200: fees = r3.json() result["fee_fast"] = fees.get("fastestFee", 0) result["fee_hour"] = fees.get("halfHourFee", 0) result["fee_economy"] = fees.get("economyFee", 0) except: pass return result def _onchain_mempool_state(self): """Current mempool state — blocks pending, congestion level.""" try: r = requests.get("https://mempool.space/api/v1/mempool-summary", timeout=8) if r.status_code == 200: data = r.json() return { "vsize_mb": round(data.get("total_vsize", 0) / 1e6, 1), "tx_pending": data.get("tx_count", 0), } except: pass return {} def _arkham_trace(self): """Arkham Intelligence — track institutional large wallets (public data).""" # Arkham has a free public API for entity tags try: r = requests.get( "https://api.arkhamintelligence.com/intelligence/labels?chain=BTC&limit=5", timeout=8, headers={"Accept": "application/json"} ) if r.status_code == 200: data = r.json() entities = [l.get("label", "") for l in data.get("results", [])[:5]] return {"arkham_entities": entities} except: pass return {} # ═══════════════════════════════════════════════════════ # PILLAR 2 — DERIVATIVES & LIQUIDITY # ═══════════════════════════════════════════════════════ def _derivatives_binance(self): """Funding rates + open interest from Binance futures (public).""" result = {} try: # Funding rate for BTC perpetual r = requests.get( "https://fapi.binance.com/fapi/v1/premiumIndex?symbol=BTCUSDT", timeout=8 ) if r.status_code == 200: data = r.json() result["funding_rate_pct"] = float(data.get("lastFundingRate", 0)) * 100 result["mark_price"] = float(data.get("markPrice", 0)) result["index_price"] = float(data.get("indexPrice", 0)) result["next_funding_ts"] = data.get("nextFundingTime", "") except: pass try: # Open interest r2 = requests.get( "https://fapi.binance.com/fapi/v1/openInterest?symbol=BTCUSDT", timeout=8 ) if r2.status_code == 200: oi = r2.json() btc_oi = int(oi.get("openInterest", 0)) result["open_interest_btc"] = btc_oi result["open_interest_usd"] = btc_oi * result.get("mark_price", 0) except: pass return result def _coinglass_funding(self): """CoinGlass funding rates aggregate across exchanges (public).""" try: r = requests.get( "https://open-api.coinglass.com/public/v2/funding_rate_history?symbol=BTC&exchange=Binance,OKX,Bybit&type=1", timeout=10, headers={"Accept": "application/json"} ) if r.status_code == 200: data = r.json() records = data.get("data", {}).get("history", []) if records: # Latest avg funding latest = records[-1] if records else {} return {"avg_funding_pct": latest.get("rate", 0)} except: pass return {} def _liquidations_map(self): """Estimate liquidation zones via open interest concentration. Source: Binance liquidated positions public data.""" result = {} try: r = requests.get( "https://fapi.binance.com/futures/data/globalLongShortAccountRatio?symbol=BTCUSDT&period=1h&limit=5", timeout=8 ) if r.status_code == 200: data = r.json() if data: latest = data[-1] long_ratio = float(latest.get("longAccount", 0)) short_ratio = float(latest.get("shortAccount", 0)) result["long_ratio_pct"] = round(long_ratio / (long_ratio + short_ratio + 0.001) * 100, 1) result["sentiment"] = "overbought" if long_ratio > 60 else "oversold" if long_ratio < 40 else "neutral" except: pass return result # ═══════════════════════════════════════════════════════ # PILLAR 3 — SENTIMENT & BEHAVIOR # ═══════════════════════════════════════════════════════ def _fear_greed_index(self): """Alternative.me Fear & Greed Index (free).""" try: r = requests.get("https://api.alternative.me/fng/?limit=2", timeout=8) if r.status_code == 200: data = r.json() items = data.get("data", []) if items: latest = items[0] prev = items[1] if len(items) > 1 else {} return { "fgi_value": int(latest.get("value", 50)), "fgi_class": latest.get("value_classification", "Neutral"), "fgi_change": int(latest.get("change", 0)), "fgi_prev_value": int(prev.get("value", 50)) if prev else 50, } except: pass return {} def _sentiment_cryptopanic(self): """CryptoPanic — news sentiment aggregator (free).""" try: r = requests.get( "https://cryptopanic.com/api/v1/posts/?auth_token=¤cies=BTC,ETH,SOL&kind=news", timeout=10 ) if r.status_code == 200: data = r.json() posts = data.get("results", [])[:20] pos_kw = ["bullish", "surge", "rally", "record", "growth", "high", "adoption", "buy", "soar", "peak", " ETF ", "institutional", "accumulation", "breakout"] neg_kw = ["bearish", "drop", "crash", "sell", "risk", "warn", "hack", "ban", "regulation", "reject", "loss", "fear"] pos = sum(1 for p in posts for t in [p.get("title", "")] if any(k.lower() in t.lower() for k in pos_kw)) neg = sum(1 for p in posts for t in [p.get("title", "")] if any(k.lower() in t.lower() for k in neg_kw)) return { "news_pos": pos, "news_neg": neg, "news_total": len(posts), "sentiment_label": "Predominantemente Otimista" if pos > neg + 3 else "Predominantemente Pessimista" if neg > pos + 3 else "Incerto" } except: pass return {} def _social_volume(self): """CoinGecko social stats (free tier).""" try: r = requests.get( "https://api.coingecko.com/api/v3/coins/bitcoin", params={"tickers": "false", "market_data": "true", "community_data": "true"}, timeout=10 ) if r.status_code == 200: data = r.json() comm = data.get("community_data", {}) or {} return { "twitter_followers": data.get("followers", 0), "forum_posts_24h": comm.get("forum_posts_active_24h", 0), "reddit_subscribers": comm.get("reddit_subscribers", 0), "telegram_channel_users": comm.get("telegram_channel_user_count", 0), } except: pass return {} # ═══════════════════════════════════════════════════════ # PILLAR 4 — TECHNICAL ANALYSIS # ═══════════════════════════════════════════════════════ def _technical_binance(self): """Candle-based support/resistance + RSI + MACD approximation.""" result = {} try: # 1h candles for 200 periods r = requests.get( "https://api.binance.com/api/v3/klines", params={"symbol": "BTCUSDT", "interval": "1h", "limit": 200}, timeout=10 ) if r.status_code == 200: candles = r.json() closes = [float(c[4]) for c in candles] highs = [float(c[2]) for c in candles] lows = [float(c[3]) for c in candles] vols = [float(c[5]) for c in candles] # RSI(14) approximation deltas = [closes[i] - closes[i-1] for i in range(1, len(closes))] gains = [d for d in deltas[-14:] if d > 0] losses = [-d for d in deltas[-14:] if d < 0] avg_gain = sum(gains) / 14 if gains else 0.001 avg_loss = sum(losses) / 14 if losses else 0.001 rs = avg_gain / avg_loss if avg_loss else 99 rsi = round(100 - (100 / (1 + rs)), 1) # MACD (12,26,9) ema12 = sum(closes[-12:]) / 12 ema26 = sum(closes[-26:]) / 26 macd_line = ema12 - ema26 signal = macd_line * 0.2 # simplified signal # Support/resistance via volume profile vol_index = vols.index(max(vols)) support = round(min(lows[:vol_index]) if vol_index > 0 else min(lows[-50:]), 2) resistance = round(max(highs[vol_index:]) if vol_index < len(highs) - 1 else max(highs[-50:]), 2) # 4h change change_4h = round((closes[-1] / closes[-5] - 1) * 100, 2) if len(closes) >= 5 else 0 change_24h = round((closes[-1] / closes[-25] - 1) * 100, 2) if len(closes) >= 25 else 0 change_7d = round((closes[-1] / closes[-169] - 1) * 100, 2) if len(closes) >= 169 else 0 result.update({ "price": closes[-1], "rsi": rsi, "macd": round(macd_line, 2), "macd_signal": round(signal, 2), "macd_histogram": round(macd_line - signal, 2), "support": support, "resistance": resistance, "change_4h": change_4h, "change_24h": change_24h, "change_7d": change_7d, "volume_avg_ratio": round(sum(vols[-24:]) / sum(vols[-168:]) * 100, 1) if sum(vols[-168:]) > 0 else 100, }) except Exception as e: pass return result # ═══════════════════════════════════════════════════════ # PILLAR 5 — MACROECONOMIC # ═══════════════════════════════════════════════════════ def _macro_etf_flows(self): """ETF flow estimates via Farside Investors (public).""" try: r = requests.get("https://farside.io/bitcoin-etf-flow-all-data", timeout=10) if r.status_code == 200: text = r.text.lower() # Look for latest date pattern and inflow/outflow indicator import re # Simple pattern: find most recent date and flow amount matches = re.findall(r'(\d{4}-\d{2}-\d{2})[^$]*?(-?\$?[\d,]+)\s*(million|billion)?\s*(inflow|outflow)', text) if matches: latest = matches[-1] return {"etf_date": latest[0], "etf_flow_raw": latest[1] + " " + (latest[2] or "")} except: pass return {} def _macro_btc_dominance(self): """BTC Dominance (Cap) — tracks altcoin season vs BTC season.""" try: r = requests.get( "https://api.coingecko.com/api/v3/global", timeout=10 ) if r.status_code == 200: data = r.json() global_data = data.get("data", {}) btc_d = global_data.get("market_cap_percentage", {}).get("btc", 0) return { "btc_dominance_pct": btc_d, "active_cryptos": global_data.get("active_cryptocurrencies", 0), } except: pass return {} def _macro_correlations(self): """DXY Dollar Index via public API.""" try: r = requests.get( "https://api.coingecko.com/api/v3/coins/us-dtate", params={"id": "us-dtate"}, timeout=8 ) # Fallback: approximate via BTC correlation with NASDAQ # Use a simple proxy from Binance DXY futures pass except: pass return {} # ═══════════════════════════════════════════════════════ # MAIN RUN # ═══════════════════════════════════════════════════════ def run(self): all_data = {} # Gather all pillars pillars = { "On-Chain": self._onchain_bitflyer, "Mempool": self._onchain_mempool_state, "Arkham": self._arkham_trace, "Derivatives": self._derivatives_binance, "Coinglass": self._coinglass_funding, "Liquidations": self._liquidations_map, "FearGreed": self._fear_greed_index, "Sentiment": self._sentiment_cryptopanic, "Social": self._social_volume, "Technical": self._technical_binance, "MacroETF": self._macro_etf_flows, "Dominance": self._macro_btc_dominance, } for name, fn in pillars.items(): try: result = fn() if result: all_data[name] = result except Exception as e: all_data[name] = {"error": str(e)} tech = all_data.get("Technical", {}) price = tech.get("price", 0) if not price: return {"summary": "Erro: não foi possível obter dados de mercado", "signal": "ERROR", "confidence": 0} # Build briefing string for Decio fgi = all_data.get("FearGreed", {}) sentiment_data = all_data.get("Sentiment", {}) deriv = all_data.get("Derivatives", {}) liq = all_data.get("Liquidations", {}) onchain = all_data.get("On-Chain", {}) btc_d = all_data.get("Dominance", {}) # Overall sentiment combining Fear&Greed + news fg_class = fgi.get("fgi_class", "Neutral") news_sent = sentiment_data.get("sentiment_label", "Incerto") funding = deriv.get("funding_rate_pct", 0) long_ratio = liq.get("long_ratio_pct", 50) btc_dom = btc_d.get("btc_dominance_pct", 0) rsi = tech.get("rsi", 50) macd_h = tech.get("macd_histogram", 0) change_4h = tech.get("change_4h", 0) # Signal detection bullish_signals = 0 bearish_signals = 0 # RSI — granular zones if rsi < 38: bullish_signals += 1 elif rsi > 62: bearish_signals += 1 # Funding rate — any positive funding = long pressure if funding > 0.0001: bearish_signals += 1 elif funding < -0.0001: bullish_signals += 1 # Long ratio — tighter around 50 if long_ratio > 53: bearish_signals += 1 elif long_ratio < 47: bullish_signals += 1 # Fear&Greed — extreme gets 2x if fg_class in ["Extreme Fear"]: bullish_signals += 2 elif fg_class == "Fear": bullish_signals += 1 elif fg_class == "Greed": bearish_signals += 1 elif fg_class in ["Extreme Greed"]: bearish_signals += 2 # MACD histogram — lower threshold for sensitivity if macd_h > 5: bullish_signals += 1 elif macd_h < -5: bearish_signals += 1 # BTC dominance if btc_dom > 55: bullish_signals += 1 elif btc_dom < 45: bearish_signals += 1 # Price momentum (4h change) if change_4h > 1.0: bullish_signals += 1 elif change_4h < -1.0: bearish_signals += 1 # Determine signal and confidence net = bullish_signals - bearish_signals if net >= 3: signal = "ALTA" confidence = min(50 + net * 8, 90) elif net <= -3: signal = "BAIXA" confidence = min(50 + abs(net) * 8, 90) else: signal = "NEUTRO" confidence = 40 + abs(net) * 5 # Change-based confirmation change_24h = tech.get("change_24h", 0) if change_24h > 5 and signal == "NEUTRO": signal = "ALTA"; confidence = min(confidence + 10, 85) elif change_24h < -5 and signal == "NEUTRO": signal = "BAIXA"; confidence = min(confidence + 10, 85) # 3-point briefing for Decio support = tech.get("support", 0) resistance = tech.get("resistance", 0) sup_str = f"${support:,.0f}" if support else "?" res_str = f"${resistance:,.0f}" if resistance else "?" briefing = ( f"BTC ${price:,.0f} | " f"4h: {change_4h:+.1f}% | 24h: {change_24h:+.1f}% | " f"RSI: {rsi} | MACD hist: {macd_h:+.2f} | " f"Sup: {sup_str} | Res: {res_str} | " f"F&G: {fg_class} ({fgi.get('fgi_value', 0)}) | " f"News: {news_sent} | " f"Funding: {funding:+.4f}% | " f"LongRatio: {long_ratio}% | " f"BTC Dom: {btc_dom}% | " f"Bullish={bullish_signals} Bearish={bearish_signals} | " f"Sinal: {signal} ({confidence}% conf)" ) result = { "price": price, "signal": signal, "confidence": confidence, "summary": f"BTC ${price:,.0f} | {signal} | conf {confidence}% | RSI {rsi} | F&G {fg_class}", "briefing_for_decio": briefing, "timestamp": datetime.now().isoformat(), # Full data for Audit "_all_pillars": all_data, # Key metrics for display "rsi": rsi, "macd_histogram": macd_h, "support": support, "resistance": resistance, "change_24h": change_24h, "change_4h": change_4h, "fear_greed": fgi.get("fgi_value", 0), "fear_greed_class": fg_class, "funding_rate": funding, "long_ratio": long_ratio, "news_sentiment": news_sent, "btc_dominance": btc_dom, "bullish_signals": bullish_signals, "bearish_signals": bearish_signals, "net_signal": net, } return result