Spaces:
Sleeping
Sleeping
File size: 4,770 Bytes
380b4bf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | #!/usr/bin/env python3
"""
RapidAgentClient - API Terminal
API para análisis de mercado. Diseñado para ser consumido por otros Spaces.
"""
from flask import Flask, request, jsonify
import os
import requests
from dotenv import load_dotenv
load_dotenv()
import yfinance as yf
import pandas as pd
import numpy as np
app = Flask(__name__)
HF_TOKEN = os.getenv("HF_TOKEN", "")
COINGECKO_API = "https://api.coingecko.com/api/v3"
SYMBOL_MAP = {
'BTC': 'bitcoin', 'ETH': 'ethereum', 'SOL': 'solana',
'ADA': 'cardano', 'DOT': 'polkadot', 'AVAX': 'avalanche-2',
'MATIC': 'matic-network', 'LINK': 'chainlink', 'XRP': 'ripple',
'DOGE': 'dogecoin', 'BNB': 'binancecoin', 'LTC': 'litecoin',
}
@app.route("/", methods=["GET"])
def index():
return jsonify({"success": True, "message": "RapidAgentClient API running"})
@app.route("/health", methods=["GET"])
def health():
return jsonify({"success": True, "status": "healthy"})
@app.route("/analyze", methods=["POST"])
def analyze():
data = request.get_json()
symbol = data.get("symbol", "BTC").upper()
price_data = get_price_data(symbol)
tech_data = get_technical_data(symbol)
analysis = generate_analysis(symbol, price_data, tech_data)
return jsonify({
"success": True,
"data": {
"symbol": symbol,
"price_data": price_data,
"tech_data": tech_data,
"analysis": analysis
}
})
def get_price_data(symbol):
try:
coin_id = SYMBOL_MAP.get(symbol, symbol.lower())
url = f"{COINGECKO_API}/coins/{coin_id}"
params = {'localization': 'false', 'tickers': 'false', 'community_data': 'false', 'developer_data': 'false', 'sparkline': 'false'}
response = requests.get(url, params=params, timeout=15)
if response.status_code == 200:
data = response.json()
return {
'success': True,
'price': data.get('market_data', {}).get('current_price', {}).get('usd', 0),
'change_24h': data.get('market_data', {}).get('price_change_percentage_24h', 0),
'change_7d': data.get('market_data', {}).get('price_change_percentage_7d', 0),
'change_30d': data.get('market_data', {}).get('price_change_percentage_30d', 0),
'market_cap': data.get('market_data', {}).get('market_cap', {}).get('usd', 0),
'volume_24h': data.get('market_data', {}).get('total_volume', {}).get('usd', 0),
'rank': data.get('market_cap_rank', 0),
}
except:
pass
return {'success': False}
def get_technical_data(symbol):
try:
ticker = yf.Ticker(f"{symbol}-USD")
hist = ticker.history(period="1mo")
if hist.empty:
return {'success': False}
current_price = hist['Close'].iloc[-1]
hist['MA7'] = hist['Close'].rolling(window=7).mean()
hist['MA20'] = hist['Close'].rolling(window=20).mean()
delta = hist['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
hist['RSI'] = 100 - (100 / (1 + rs))
volatility = hist['Close'].pct_change().std() * 100
return {
'success': True,
'price': float(current_price),
'rsi': float(hist['RSI'].iloc[-1]) if not pd.isna(hist['RSI'].iloc[-1]) else 50.0,
'ma7': float(hist['MA7'].iloc[-1]) if not pd.isna(hist['MA7'].iloc[-1]) else float(current_price),
'ma20': float(hist['MA20'].iloc[-1]) if not pd.isna(hist['MA20'].iloc[-1]) else float(current_price),
'volatility': float(volatility * 100),
'trend': 'bullish' if current_price > hist['MA7'].iloc[-1] else 'bearish' if current_price < hist['MA7'].iloc[-1] else 'neutral',
}
except:
return {'success': False}
def generate_analysis(symbol, price_data, tech_data):
rsi = tech_data.get('rsi', 50)
trend = tech_data.get('trend', 'neutral')
change_24h = price_data.get('change_24h', 0)
if rsi > 70:
signal = "vender"
risk = "alto"
elif rsi < 30:
signal = "comprar"
risk = "medio"
elif trend == "bullish" and change_24h > 0:
signal = "comprar"
risk = "bajo"
elif trend == "bearish" and change_24h < 0:
signal = "vender"
risk = "medio"
else:
signal = "mantener"
risk = "bajo"
return f'{{"tendencia": "{trend}", "senal": "{signal}", "riesgo": "{risk}", "accion": "Considerar {signal} {symbol}"}}'
if __name__ == "__main__":
app.run(host="0.0.0.0", port=7860)
|