---
async function analyzeBitcoin() {
const response = await fetch('http://localhost:8000/api/analyze', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({
crypto_id: 'bitcoin',
github_owner: 'bitcoin',
github_repo: 'bitcoin',
subreddit: 'Bitcoin'
})
});
const data = await response.json();
console.log(Bubble Index: ${data.metrics.bubble_index});
console.log(Risk Level: ${data.metrics.risk_level});
return data;
}
analyzeBitcoin();
import axios from 'axios';
const api = axios.create({
baseURL: 'http://localhost:8000',
timeout: 30000
});
async function getHistoricalData(asset, days = 90) {
try {
const { data } = await api.get(/api/historical/${asset}, {
params: { days }
});
return data;
} catch (error) {
console.error('Error:', error.response?.data || error.message);
throw error;
}
}
// Uso
const history = await getHistoricalData('bitcoin', 90);
console.log(${history.count} measurements in ${history.days} days);
import { useState, useEffect } from 'react';
function useBubbleAnalysis(config) {
const [data, setData] = useState(null);
const [loading, setLoading] = useState(false);
const [error, setError] = useState(null);
useEffect(() => {
const fetchData = async () => {
setLoading(true);
setError(null);
try {
const response = await fetch('http://localhost:8000/api/analyze', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(config)
});
if (!response.ok) throw new Error('API Error');
const result = await response.json();
setData(result);
} catch (err) {
setError(err.message);
} finally {
setLoading(false);
}
};
if (config) fetchData();
}, [config]);
return { data, loading, error };
}
// Uso no componente
function BubbleAnalyzer() {
const { data, loading, error } = useBubbleAnalysis({
crypto_id: 'bitcoin',
github_owner: 'bitcoin',
github_repo: 'bitcoin',
subreddit: 'Bitcoin'
});
if (loading) return Loading...;
if (error) return Error: {error};
if (!data) return null;
return (
Bubble Index: {(data.metrics.bubble_index * 100).toFixed(1)}%
Risk: {data.metrics.risk_level}
);
}
---
import requests
from typing import Dict, Optional
class BubbleDetectorClient:
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url
self.session = requests.Session()
def analyze(
self,
crypto_id: Optional[str] = None,
github_owner: Optional[str] = None,
github_repo: Optional[str] = None,
subreddit: Optional[str] = None,
npm_package: Optional[str] = None,
force_refresh: bool = False
) -> Dict:
"""Analisa um ativo"""
url = f"{self.base_url}/api/analyze"
params = {"force_refresh": force_refresh}
data = {
"crypto_id": crypto_id,
"github_owner": github_owner,
"github_repo": github_repo,
"subreddit": subreddit,
"npm_package": npm_package
}
# Remover valores None
data = {k: v for k, v in data.items() if v is not None}
response = self.session.post(url, json=data, params=params)
response.raise_for_status()
return response.json()
def get_historical(self, asset: str, days: int = 90) -> Dict:
"""Obtém dados históricos"""
url = f"{self.base_url}/api/historical/{asset}"
params = {"days": days}
response = self.session.get(url, params=params)
response.raise_for_status()
return response.json()
def get_presets(self) -> Dict:
"""Obtém presets disponíveis"""
url = f"{self.base_url}/api/presets"
response = self.session.get(url)
response.raise_for_status()
return response.json()
Uso
client = BubbleDetectorClient()
Analisar Bitcoin
result = client.analyze(
crypto_id="bitcoin",
github_owner="bitcoin",
github_repo="bitcoin",
subreddit="Bitcoin"
)
print(f"Bubble Index: {result['metrics']['bubble_index']:.2%}")
print(f"Risk Level: {result['metrics']['risk_level']}")
Histórico
history = client.get_historical("bitcoin", days=90)
print(f"{history['count']} measurements")
import httpx
import asyncio
from typing import List, Dict
class AsyncBubbleClient:
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url
async def analyze_multiple(self, configs: List[Dict]) -> List[Dict]:
"""Analisa múltiplos ativos em paralelo"""
async with httpx.AsyncClient(timeout=30.0) as client:
tasks = [
client.post(
f"{self.base_url}/api/analyze",
json=config
)
for config in configs
]
responses = await asyncio.gather(*tasks)
return [r.json() for r in responses]
Uso
async def main():
client = AsyncBubbleClient()
configs = [
{"crypto_id": "bitcoin", "subreddit": "Bitcoin"},
{"crypto_id": "ethereum", "subreddit": "ethereum"},
{"crypto_id": "solana", "subreddit": "solana"}
]
results = await client.analyze_multiple(configs)
for result in results:
asset = result['asset']
index = result['metrics']['bubble_index']
print(f"{asset}: {index:.2%}")
asyncio.run(main())
---
import pandas as pd
import requests
def get_bubble_dataframe(asset: str, days: int = 90) -> pd.DataFrame:
"""Retorna dados históricos como DataFrame"""
url = f"http://localhost:8000/api/historical/{asset}"
response = requests.get(url, params={"days": days})
data = response.json()
df = pd.DataFrame(data['data'])
# Converter timestamp para datetime
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('datetime', inplace=True)
return df
Análise
df = get_bubble_dataframe('bitcoin', days=90)
print("Estatísticas:")
print(df[['bubble_index', 'adoption', 'hype']].describe())
Plot
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(12, 6))
df[['bubble_index', 'adoption', 'hype']].plot(ax=ax)
ax.set_ylabel('Score (0-1)')
ax.set_title('Bitcoin - Bubble Metrics Over Time')
ax.legend()
plt.tight_layout()
plt.show()
---
use reqwest;
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize)]
struct AssetConfig {
crypto_id: Option,
github_owner: Option,
github_repo: Option,
subreddit: Option,
npm_package: Option,
}
#[derive(Debug, Deserialize)]
struct BubbleMetrics {
adoption: f64,
hype: f64,
investment: f64,
network: f64,
feedback: f64,
bubble_index: f64,
risk_level: String,
}
#[derive(Debug, Deserialize)]
struct AnalysisResponse {
asset: String,
metrics: BubbleMetrics,
cached: bool,
}
#[tokio::main]
async fn main() -> Result<(), Box> {
let client = reqwest::Client::new();
let config = AssetConfig {
crypto_id: Some("bitcoin".to_string()),
github_owner: Some("bitcoin".to_string()),
github_repo: Some("bitcoin".to_string()),
subreddit: Some("Bitcoin".to_string()),
npm_package: None,
};
let response = client
.post("http://localhost:8000/api/analyze")
.json(&config)
.send()
.await?;
let analysis: AnalysisResponse = response.json().await?;
println!("Asset: {}", analysis.asset);
println!("Bubble Index: {:.2}%", analysis.metrics.bubble_index * 100.0);
println!("Risk Level: {}", analysis.metrics.risk_level);
Ok(())
}
---
package main
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
)
type AssetConfig struct {
CryptoID string json:"crypto_id,omitempty"
GithubOwner string json:"github_owner,omitempty"
GithubRepo string json:"github_repo,omitempty"
Subreddit string json:"subreddit,omitempty"
NPMPackage string json:"npm_package,omitempty"
}
type BubbleMetrics struct {
Adoption float64 json:"adoption"
Hype float64 json:"hype"
Investment float64 json:"investment"
Network float64 json:"network"
Feedback float64 json:"feedback"
BubbleIndex float64 json:"bubble_index"
RiskLevel string json:"risk_level"
}
type AnalysisResponse struct {
Asset string json:"asset"
Metrics BubbleMetrics json:"metrics"
Cached bool json:"cached"
}
func analyzeBitcoin() (*AnalysisResponse, error) {
config := AssetConfig{
CryptoID: "bitcoin",
GithubOwner: "bitcoin",
GithubRepo: "bitcoin",
Subreddit: "Bitcoin",
}
jsonData, err := json.Marshal(config)
if err != nil {
return nil, err
}
resp, err := http.Post(
"http://localhost:8000/api/analyze",
"application/json",
bytes.NewBuffer(jsonData),
)
if err != nil {
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return nil, err
}
var analysis AnalysisResponse
err = json.Unmarshal(body, &analysis)
if err != nil {
return nil, err
}
return &analysis, nil
}
func main() {
analysis, err := analyzeBitcoin()
if err != nil {
fmt.Printf("Error: %v\n", err)
return
}
fmt.Printf("Asset: %s\n", analysis.Asset)
fmt.Printf("Bubble Index: %.2f%%\n", analysis.Metrics.BubbleIndex*100)
fmt.Printf("Risk Level: %s\n", analysis.Metrics.RiskLevel)
}
---
curl -X POST "http://localhost:8000/api/analyze" \
-H "Content-Type: application/json" \
-d '{
"crypto_id": "bitcoin",
"github_owner": "bitcoin",
"github_repo": "bitcoin",
"subreddit": "Bitcoin"
}'
curl -X POST "http://localhost:8000/api/analyze?force_refresh=true" \
-H "Content-Type: application/json" \
-d '{
"crypto_id": "ethereum",
"subreddit": "ethereum"
}'
curl "http://localhost:8000/api/historical/bitcoin?days=90"
curl -s "http://localhost:8000/api/analyze" \
-H "Content-Type: application/json" \
-d '{"crypto_id": "bitcoin"}' | jq '.metrics'
---
{
"info": {
"name": "Tech Bubble Detector API",
"schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
},
"item": [
{
"name": "Analyze Bitcoin",
"request": {
"method": "POST",
"header": [
{
"key": "Content-Type",
"value": "application/json"
}
],
"body": {
"mode": "raw",
"raw": "{\n \"crypto_id\": \"bitcoin\",\n \"github_owner\": \"bitcoin\",\n \"github_repo\": \"bitcoin\",\n \"subreddit\": \"Bitcoin\"\n}"
},
"url": {
"raw": "http://localhost:8000/api/analyze",
"protocol": "http",
"host": ["localhost"],
"port": "8000",
"path": ["api", "analyze"]
}
}
},
{
"name": "Get Historical Data",
"request": {
"method": "GET",
"url": {
"raw": "http://localhost:8000/api/historical/bitcoin?days=90",
"protocol": "http",
"host": ["localhost"],
"port": "8000",
"path": ["api", "historical", "bitcoin"],
"query": [
{
"key": "days",
"value": "90"
}
]
}
}
}
]
}
---
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import requests
def analyze_top_cryptos(**context):
"""Analisa top 10 cryptos e armazena resultados"""
cryptos = ['bitcoin', 'ethereum', 'solana', 'cardano', 'polkadot']
results = []
for crypto in cryptos:
response = requests.post(
'http://bubble-api:8000/api/analyze',
json={'crypto_id': crypto, 'subreddit': crypto}
)
if response.ok:
data = response.json()
results.append({
'crypto': crypto,
'bubble_index': data['metrics']['bubble_index'],
'risk_level': data['metrics']['risk_level']
})
# Push para XCom
context['task_instance'].xcom_push(key='analysis_results', value=results)
return results
def generate_report(**context):
"""Gera relatório baseado nos resultados"""
results = context['task_instance'].xcom_pull(
task_ids='analyze_cryptos',
key='analysis_results'
)
# Gerar relatório
high_risk = [r for r in results if r['bubble_index'] > 0.7]
if high_risk:
print(f"ALERTA: {len(high_risk)} cryptos em alto risco!")
for crypto in high_risk:
print(f" - {crypto['crypto']}: {crypto['bubble_index']:.2%}")
default_args = {
'owner': 'data-team',
'depends_on_past': False,
'start_date': datetime(2024, 1, 1),
'email_on_failure': True,
'email_on_retry': False,
'retries': 2,
'retry_delay': timedelta(minutes=5),
}
dag = DAG(
'bubble_detector_daily',
default_args=default_args,
description='Análise diária de bolhas tecnológicas',
schedule_interval='0 9 * * *', # Diariamente às 9h
catchup=False
)
t1 = PythonOperator(
task_id='analyze_cryptos',
python_callable=analyze_top_cryptos,
dag=dag,
)
t2 = PythonOperator(
task_id='generate_report',
python_callable=generate_report,
dag=dag,
)
t1 >> t2
---
Adicione ao backend_api.py
from prometheus_client import Counter, Histogram, Gauge, generate_latest
Métricas
api_requests_total = Counter(
'api_requests_total',
'Total API requests',
['method', 'endpoint', 'status']
)
bubble_index_gauge = Gauge(
'bubble_index',
'Current bubble index',
['asset']
)
response_time_histogram = Histogram(
'response_time_seconds',
'Response time in seconds',
['endpoint']
)
@app.get("/metrics")
async def metrics():
return Response(generate_latest(), media_type="text/plain")
scrape_configs:
- job_name: 'bubble-detector'
static_configs:
- targets: ['localhost:8000']
metrics_path: '/metrics'
scrape_interval: 30s
---
import requests
def send_slack_alert(asset: str, bubble_index: float, risk_level: str):
"""Envia alerta para Slack"""
webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
color = "#10b981" if bubble_index < 0.3 else \
"#f59e0b" if bubble_index < 0.5 else \
"#ef4444" if bubble_index < 0.7 else "#991b1b"
message = {
"attachments": [{
"color": color,
"title": f"Bubble Alert: {asset}",
"fields": [
{
"title": "Bubble Index",
"value": f"{bubble_index:.2%}",
"short": True
},
{
"title": "Risk Level",
"value": risk_level,
"short": True
}
],
"footer": "Tech Bubble Detector",
"ts": int(datetime.now().timestamp())
}]
}
requests.post(webhook_url, json=message)
Uso
if bubble_index > 0.7:
send_slack_alert('bitcoin', bubble_index, 'Crítico')
---
Estes exemplos cobrem as principais linguagens e ferramentas de integração. Para mais detalhes, consulte a documentação completa.