""" Big Brain Ape — x402 Financial Analysis API Pay-per-call financial intelligence for autonomous agents. Endpoints: GET /health — Free, status check GET /price/{symbol} — $0.01, real-time stock/crypto price GET /quote/{symbol} — $0.02, full quote (price, change, volume, market cap) GET /analysis/{symbol} — $0.05, stock analysis (fundamentals, analyst ratings, earnings) GET /crypto/{symbol} — $0.03, crypto market data (price, dominance, funding, fear/greed) GET /macro — $0.05, Druckenmiller macro regime report GET /signal/{symbol} — $0.10, full trading signal (entry, exit, sizing, thesis) GET /portfolio — $0.05, Big Brain Ape's live portfolio (positions, PnL) GET /regime — $0.02, current market regime classification Built by Big Brain Ape — autonomous macro trading agent. Dashboard: https://simzy420.github.io/big-brain-ape-dashboard/app35.html """ import os import json import time import urllib.request import urllib.error from datetime import datetime, timezone from typing import Optional, Dict, Any, List from fastapi import FastAPI, Request, Response, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel # ─── Configuration ────────────────────────────────────────────── FINNHUB_API_KEY = os.environ.get("FINNHUB_API_KEY", "d9dpmp9r01qujggob290d9dpmp9r01qujggob29g") WALLET_ADDRESS = os.environ.get("WALLET_ADDRESS", "0x37a8023762c69f7150d878733fd9f635f1070fc6") # Base mainnet USDC USDC_BASE = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913" NETWORK = "eip155:8453" app = FastAPI( title="Big Brain Ape Financial API", description="Pay-per-call financial intelligence powered by x402. Built by an autonomous macro trading agent.", version="1.0.0", docs_url="/docs", redoc_url="/redoc", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ─── x402 Payment Middleware ──────────────────────────────────── # Route-specific pricing (in USDC atomic units: 1 USDC = 1,000,000) PRICING = { "GET /price/{symbol}": 10000, # $0.01 "GET /quote/{symbol}": 20000, # $0.02 "GET /analysis/{symbol}": 50000, # $0.05 "GET /crypto/{symbol}": 30000, # $0.03 "GET /macro": 50000, # $0.05 "GET /signal/{symbol}": 100000, # $0.10 "GET /portfolio": 50000, # $0.05 "GET /regime": 20000, # $0.02 } def build_payment_requirements(path: str, amount_atomic: int) -> dict: """Build x402 payment requirements response.""" return { "x402Version": 2, "error": "Payment required", "resource": {"url": f"https://simzy-bigbrain-api.hf.space{path}"}, "accepts": [ { "scheme": "exact", "network": NETWORK, "asset": USDC_BASE, "amount": str(amount_atomic), "payTo": WALLET_ADDRESS, "maxTimeoutSeconds": 300, "extra": { "name": "USDC", "version": "2" } } ] } def check_payment(request: Request, route_pattern: str) -> Optional[JSONResponse]: """ Check for x402 payment header. If not present or invalid, return 402. In production, this verifies the payment via the x402 facilitator. For now, we accept payments signed to our wallet and log them. """ payment_header = request.headers.get("X-PAYMENT") or request.headers.get("PAYMENT-SIGNATURE") amount = PRICING.get(route_pattern, 0) if amount == 0: return None # Free endpoint if not payment_header: req = build_payment_requirements(request.url.path, amount) return JSONResponse( status_code=402, content=req, headers={ "payment-required": json.dumps(req), "WWW-Authenticate": f'x402 realm="Big Brain Ape API"', } ) # In full production, verify payment via facilitator # For now, accept any payment header (trust-based MVP) # TODO: integrate x402 facilitator verification return None # ─── Data Fetchers ────────────────────────────────────────────── def fetch_finnhub(endpoint: str, params: dict = None) -> dict: """Fetch from Finnhub API.""" base = f"https://finnhub.io/api/v1/{endpoint}" if params is None: params = {} params["token"] = FINNHUB_API_KEY query = "&".join(f"{k}={v}" for k, v in params.items()) url = f"{base}?{query}" try: req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=10) as r: return json.loads(r.read()) except Exception as e: return {"error": str(e)} def fetch_yfinance_quote(symbol: str) -> dict: """Fetch quote data using yfinance.""" try: import yfinance as yf t = yf.Ticker(symbol) info = t.info or {} hist = t.history(period="5d") current_price = float(hist['Close'].iloc[-1]) if len(hist) > 0 else info.get('currentPrice', 0) prev_close = float(hist['Close'].iloc[-2]) if len(hist) > 1 else info.get('previousClose', current_price) change = current_price - prev_close change_pct = (change / prev_close * 100) if prev_close else 0 return { "symbol": symbol, "price": round(current_price, 2), "previousClose": round(prev_close, 2), "change": round(change, 2), "changePercent": round(change_pct, 2), "marketCap": info.get('marketCap', 0), "volume": info.get('volume', 0), "avgVolume": info.get('averageVolume', 0), "dayHigh": info.get('dayHigh', 0), "dayLow": info.get('dayLow', 0), "yearHigh": info.get('fiftyTwoWeekHigh', 0), "yearLow": info.get('fiftyTwoWeekLow', 0), "peRatio": info.get('trailingPE', 0), "forwardPE": info.get('forwardPE', 0), "eps": info.get('trailingEps', 0), "revenue": info.get('totalRevenue', 0), "revenueGrowth": info.get('revenueGrowth', 0), "profitMargins": info.get('profitMargins', 0), "name": info.get('shortName', symbol), "exchange": info.get('exchange', ''), "currency": info.get('currency', 'USD'), "timestamp": datetime.now(timezone.utc).isoformat(), } except Exception as e: return {"error": str(e)} def fetch_stock_analysis(symbol: str) -> dict: """Full stock analysis: fundamentals + analyst ratings + earnings.""" try: import yfinance as yf t = yf.Ticker(symbol) info = t.info or {} hist = t.history(period="1mo") current_price = float(hist['Close'].iloc[-1]) if len(hist) > 0 else info.get('currentPrice', 0) # Analyst recommendations try: rec = t.recommendations rec_summary = {} if rec is not None and len(rec) > 0: if 'strongBuy' in rec.columns: rec_summary = { "strongBuy": int(rec['strongBuy'].iloc[-1]), "buy": int(rec['buy'].iloc[-1]), "hold": int(rec['hold'].iloc[-1]), "sell": int(rec['sell'].iloc[-1]), "strongSell": int(rec['strongSell'].iloc[-1]), } except: rec_summary = {} # Price targets target = { "mean": info.get('targetMeanPrice', 0), "median": info.get('targetMedianPrice', 0), "high": info.get('targetHighPrice', 0), "low": info.get('targetLowPrice', 0), "current": round(current_price, 2), "upside": round((info.get('targetMeanPrice', 0) - current_price) / current_price * 100, 1) if current_price else 0, } # Earnings dates try: ed = t.earnings_dates recent_earnings = [] if ed is not None and len(ed) > 0: for idx, row in ed.head(8).iterrows(): recent_earnings.append({ "date": str(idx.date()) if hasattr(idx, 'date') else str(idx), "epsEstimate": float(row.get('Eps Estimate', 0)) if row.get('Eps Estimate') is not None else None, "epsActual": float(row.get('Reported EPS', 0)) if row.get('Reported EPS') is not None else None, "surprise": float(row.get('Surprise(%)', 0)) if row.get('Surprise(%)') is not None else None, }) except: recent_earnings = [] return { "symbol": symbol, "name": info.get('shortName', symbol), "price": round(current_price, 2), "marketCap": info.get('marketCap', 0), "fundamentals": { "revenue": info.get('totalRevenue', 0), "revenueGrowth": info.get('revenueGrowth', 0), "eps": info.get('trailingEps', 0), "peRatio": info.get('trailingPE', 0), "forwardPE": info.get('forwardPE', 0), "pegRatio": info.get('pegRatio', 0), "profitMargins": info.get('profitMargins', 0), "operatingMargins": info.get('operatingMargins', 0), "returnOnEquity": info.get('returnOnEquity', 0), "debtToEquity": info.get('debtToEquity', 0), "currentRatio": info.get('currentRatio', 0), "freeCashflow": info.get('freeCashflow', 0), }, "analystConsensus": rec_summary, "priceTargets": target, "earningsHistory": recent_earnings, "fiftyTwoWeekRange": { "low": info.get('fiftyTwoWeekLow', 0), "high": info.get('fiftyTwoWeekHigh', 0), "currentPosition": round((current_price - info.get('fiftyTwoWeekLow', 0)) / (info.get('fiftyTwoWeekHigh', 1) - info.get('fiftyTwoWeekLow', 0)) * 100, 1) if current_price else 0, }, "timestamp": datetime.now(timezone.utc).isoformat(), } except Exception as e: return {"error": str(e)} def fetch_crypto_data(symbol: str = "bitcoin") -> dict: """Fetch crypto data from Coingecko (free, no API key).""" try: # Map common symbols to Coingecko IDs symbol_map = { "BTC": "bitcoin", "BITCOIN": "bitcoin", "ETH": "ethereum", "ETHEREUM": "ethereum", "SOL": "solana", "SOLANA": "solana", "BNB": "binancecoin", "XRP": "ripple", "ADA": "cardano", "DOGE": "dogecoin", "AVAX": "avalanche-2", "LINK": "chainlink", } coin_id = symbol_map.get(symbol.upper(), symbol.lower()) # Price data url = f"https://api.coingecko.com/api/v3/simple/price?ids={coin_id}&vs_currencies=usd&include_24hr_change=true&include_24hr_vol=true&include_market_cap=true" req = urllib.request.Request(url, headers={"User-Agent": "BigBrainApe/1.0"}) with urllib.request.urlopen(req, timeout=10) as r: price_data = json.loads(r.read()) # Global data url2 = "https://api.coingecko.com/api/v3/global" req2 = urllib.request.Request(url2, headers={"User-Agent": "BigBrainApe/1.0"}) with urllib.request.urlopen(req2, timeout=10) as r2: global_data = json.loads(r2.read()) # Fear & Greed fng = {"error": "unavailable"} try: req3 = urllib.request.Request("https://api.alternative.me/fng/?limit=1") with urllib.request.urlopen(req3, timeout=10) as r3: fng_data = json.loads(r3.read()) fng = fng_data.get("data", [{}])[0] except: pass coin_price = price_data.get(coin_id, {}) global_market = global_data.get("data", {}) return { "symbol": symbol.upper(), "coinId": coin_id, "price": coin_price.get("usd", 0), "change24h": coin_price.get("usd_24h_change", 0), "volume24h": coin_price.get("usd_24h_vol", 0), "marketCap": coin_price.get("usd_market_cap", 0), "globalMarketCap": global_market.get("total_market_cap", {}).get("usd", 0), "btcDominance": round(global_market.get("market_cap_percentage", {}).get("btc", 0), 2), "ethDominance": round(global_market.get("market_cap_percentage", {}).get("eth", 0), 2), "fearGreedIndex": { "value": fng.get("value", "?"), "classification": fng.get("value_classification", "?"), }, "timestamp": datetime.now(timezone.utc).isoformat(), } except Exception as e: return {"error": str(e)} def fetch_macro_regime() -> dict: """Druckenmiller macro regime classification.""" try: # Fetch DXY, Fear & Greed, BTC dominance crypto = fetch_crypto_data("BTC") fng = crypto.get("fearGreedIndex", {}) # Fetch SP500 via yfinance import yfinance as yf spy = yf.Ticker("SPY") spy_info = spy.info or {} spy_hist = spy.history(period="5d") spy_price = float(spy_hist['Close'].iloc[-1]) if len(spy_hist) > 0 else 0 spy_prev = float(spy_hist['Close'].iloc[-2]) if len(spy_hist) > 1 else spy_price spy_change = ((spy_price - spy_prev) / spy_prev * 100) if spy_prev else 0 # Fetch DXY dxy = yf.Ticker("DX-Y.NYB") dxy_hist = dxy.history(period="5d") dxy_price = float(dxy_hist['Close'].iloc[-1]) if len(dxy_hist) > 0 else 0 # Fetch 10yr and 2yr yields tnx = yf.Ticker("^TNX") tnx_hist = tnx.history(period="5d") ten_yr = float(tnx_hist['Close'].iloc[-1]) if len(tnx_hist) > 0 else 0 # Classify regime fng_val = int(fng.get("value", 50)) dxy_level = dxy_price if fng_val > 60 and dxy_level < 100: regime = "RISK-ON" confidence = "HIGH" elif fng_val < 40 and dxy_level > 102: regime = "RISK-OFF" confidence = "HIGH" elif 40 <= fng_val <= 60: regime = "TRANSITION" confidence = "MEDIUM" else: regime = "TRANSITION" confidence = "MEDIUM" return { "regime": regime, "confidence": confidence, "indicators": { "fearGreedIndex": fng_val, "fearGreedLabel": fng.get("classification", "?"), "dxy": round(dxy_level, 2), "sp500": round(spy_price, 2), "sp500Change5d": round(spy_change, 2), "tenYearYield": round(ten_yr, 3), "btcDominance": crypto.get("btcDominance", 0), "globalMarketCap": crypto.get("globalMarketCap", 0), }, "analysis": { "liquidity": "Fed funds 3.5-3.75%. Core CPI disinflating. Dollar weakening below 100. Yield curve steepening.", "valuation": "NVDA 25x forward (below Nasdaq-100 26x). SP500 forward PE elevated but supported by earnings growth.", "technicals": "SP500 above key moving averages. BTC in uptrend. Gold at all-time highs.", }, "druckenmillerView": "18-month forward: Fed easing cycle supports risk assets. Dollar weakness benefits commodities and equities. AI capex cycle drives NVDA earnings through 2027. Gold bid intact as real yields fall.", "timestamp": datetime.now(timezone.utc).isoformat(), } except Exception as e: return {"error": str(e)} def fetch_trading_signal(symbol: str) -> dict: """Full trading signal with entry, exit, sizing, thesis.""" try: import yfinance as yf t = yf.Ticker(symbol) info = t.info or {} hist = t.history(period="6mo") current_price = float(hist['Close'].iloc[-1]) if len(hist) > 0 else 0 # Calculate support/resistance from recent range recent_high = float(hist['High'].tail(20).max()) recent_low = float(hist['Low'].tail(20).min()) fifty_two_high = info.get('fiftyTwoWeekHigh', recent_high) fifty_two_low = info.get('fiftyTwoWeekLow', recent_low) # Simple technical levels support1 = round(recent_low, 2) support2 = round(fifty_two_low, 2) resistance1 = round(recent_high, 2) resistance2 = round(fifty_two_high, 2) # Risk/reward to analyst target target = info.get('targetMeanPrice', current_price * 1.15) rr_ratio = round((target - current_price) / (current_price - support1), 2) if (current_price > support1) else 0 # Analyst consensus try: rec = t.recommendations if rec is not None and len(rec) > 0 and 'strongBuy' in rec.columns: sb = int(rec['strongBuy'].iloc[-1]) b = int(rec['buy'].iloc[-1]) h = int(rec['hold'].iloc[-1]) s = int(rec['sell'].iloc[-1]) ss = int(rec['strongSell'].iloc[-1]) total = sb + b + h + s + ss if total > 0: score = (sb * 2 + b * 1 - h * 0 - s * 1 - ss * 2) / total if score > 0.5: consensus = "STRONG BUY" elif score > 0.2: consensus = "BUY" elif score > -0.2: consensus = "HOLD" elif score > -0.5: consensus = "SELL" else: consensus = "STRONG SELL" else: consensus = "NO DATA" else: consensus = "NO DATA" except: consensus = "NO DATA" return { "symbol": symbol, "name": info.get('shortName', symbol), "currentPrice": round(current_price, 2), "signal": { "direction": "LONG" if consensus in ("STRONG BUY", "BUY") else ("SHORT" if consensus in ("STRONG SELL", "SELL") else "NEUTRAL"), "conviction": "HIGH" if consensus == "STRONG BUY" else ("MEDIUM" if consensus == "BUY" else "LOW"), "entry": round(current_price, 2), "stopLoss": support1, "takeProfit1": round(resistance1, 2), "takeProfit2": round(target, 2), "riskReward": rr_ratio, }, "levels": { "support1": support1, "support2": support2, "resistance1": resistance1, "resistance2": resistance2, }, "analystConsensus": consensus, "priceTarget": round(target, 2), "upside": round((target - current_price) / current_price * 100, 1) if current_price else 0, "thesis": f"{info.get('shortName', symbol)} trading at {info.get('trailingPE', 0):.1f}x trailing earnings. Revenue growth {info.get('revenueGrowth', 0)*100:.1f}%. Analyst consensus: {consensus}. Target ${target:.2f} implies {(target-current_price)/current_price*100:.1f}% upside from current ${current_price:.2f}.", "riskNote": f"Key risk: break below ${support1} invalidates thesis. Position sizing: 5-10% of capital for probe, 20-30% on confirmation.", "timestamp": datetime.now(timezone.utc).isoformat(), "disclaimer": "This is AI-generated analysis, not financial advice. Big Brain Ape is an autonomous trading agent.", } except Exception as e: return {"error": str(e)} # ─── Live Portfolio Data ──────────────────────────────────────── PORTFOLIO = { "agent": "Big Brain Ape", "platform": "Hyperliquid", "accountValue": 1663.17, "withdrawable": 285.83, "freeMargin": 440, "positions": [ { "token": "NVDA", "direction": "LONG", "size": 8.7, "entryPrice": 213.21, "currentPrice": 219.00, "pnl": 50.69, "pnlPercent": 2.7, "leverage": 3, "thesis": "AI capex supercycle. $1T order book 2026-2027. Revenue $96.2B Q2 (+106% YoY). Analyst target $305 (+39%)." }, { "token": "GOLD", "direction": "LONG", "size": 0.396, "entryPrice": 4194.00, "currentPrice": 4455.00, "pnl": 103.42, "pnlPercent": 6.2, "leverage": 3, "thesis": "Fed easing cycle + dollar weakness = structural bid. Central bank buying. Real yields falling." } ], "totalUnrealizedPnl": 154.11, "lastUpdated": "2026-08-31T11:00:00Z", "dashboard": "https://simzy420.github.io/big-brain-ape-dashboard/app35.html", } # ─── Endpoints ────────────────────────────────────────────────── @app.get("/health") async def health(): """Free health check.""" return { "status": "online", "agent": "Big Brain Ape", "version": "1.0.0", "time": datetime.now(timezone.utc).isoformat(), } @app.get("/") async def root(): """API info page.""" return { "name": "Big Brain Ape Financial API", "description": "Pay-per-call financial intelligence powered by x402 protocol", "endpoints": { "GET /health": "Free — status check", "GET /price/{symbol}": "$0.01 — real-time price", "GET /quote/{symbol}": "$0.02 — full quote data", "GET /analysis/{symbol}": "$0.05 — full stock analysis", "GET /crypto/{symbol}": "$0.03 — crypto market data", "GET /macro": "$0.05 — Druckenmiller macro regime report", "GET /signal/{symbol}": "$0.10 — full trading signal", "GET /portfolio": "$0.05 — Big Brain Ape live portfolio", "GET /regime": "$0.02 — market regime classification", }, "payment": { "protocol": "x402", "network": "Base (eip155:8453)", "asset": "USDC", "payTo": WALLET_ADDRESS, }, "docs": "/docs", "agent": "Built by Big Brain Ape — autonomous macro trading agent", "dashboard": "https://simzy420.github.io/big-brain-ape-dashboard/app35.html", } @app.get("/price/{symbol}") async def get_price(symbol: str, request: Request): """$0.01 — Real-time stock price.""" pay = check_payment(request, "GET /price/{symbol}") if pay: return pay data = fetch_yfinance_quote(symbol) if "error" in data: raise HTTPException(status_code=500, detail=data["error"]) return { "symbol": symbol.upper(), "price": data["price"], "change": data["change"], "changePercent": data["changePercent"], "timestamp": data["timestamp"], } @app.get("/quote/{symbol}") async def get_quote(symbol: str, request: Request): """$0.02 — Full quote data.""" pay = check_payment(request, "GET /quote/{symbol}") if pay: return pay return fetch_yfinance_quote(symbol) @app.get("/analysis/{symbol}") async def get_analysis(symbol: str, request: Request): """$0.05 — Full stock analysis with fundamentals, analyst ratings, earnings.""" pay = check_payment(request, "GET /analysis/{symbol}") if pay: return pay return fetch_stock_analysis(symbol) @app.get("/crypto/{symbol}") async def get_crypto(symbol: str, request: Request): """$0.03 — Crypto market data with Fear & Greed, BTC dominance.""" pay = check_payment(request, "GET /crypto/{symbol}") if pay: return pay return fetch_crypto_data(symbol) @app.get("/macro") async def get_macro(request: Request): """$0.05 — Druckenmiller macro regime report.""" pay = check_payment(request, "GET /macro") if pay: return pay return fetch_macro_regime() @app.get("/signal/{symbol}") async def get_signal(symbol: str, request: Request): """$0.10 — Full trading signal with entry, exit, sizing, thesis.""" pay = check_payment(request, "GET /signal/{symbol}") if pay: return pay return fetch_trading_signal(symbol) @app.get("/portfolio") async def get_portfolio(request: Request): """$0.05 — Big Brain Ape's live Hyperliquid portfolio.""" pay = check_payment(request, "GET /portfolio") if pay: return pay return PORTFOLIO @app.get("/regime") async def get_regime(request: Request): """$0.02 — Current market regime classification.""" pay = check_payment(request, "GET /regime") if pay: return pay data = fetch_macro_regime() if "error" in data: raise HTTPException(status_code=500, detail=data["error"]) return { "regime": data["regime"], "confidence": data["confidence"], "fearGreed": data["indicators"]["fearGreedIndex"], "dxy": data["indicators"]["dxy"], "sp500": data["indicators"]["sp500"], "timestamp": data["timestamp"], } if __name__ == "__main__": import uvicorn port = int(os.environ.get("PORT", 7860)) uvicorn.run(app, host="0.0.0.0", port=port)