#!/usr/bin/env python3 """ RapidLiveClient - Trading API API de trading en vivo. Diseñado para ser consumido por RapidQuant. """ from flask import Flask, request, jsonify import os import uuid from datetime import datetime import warnings warnings.filterwarnings('ignore') import yfinance as yf import numpy as np import pandas as pd import requests from bs4 import BeautifulSoup try: import tensorflow as tf from tensorflow.keras.models import Sequential from tensorflow.keras.layers import LSTM, Dense, Dropout from tensorflow.keras.optimizers import Adam TF_AVAILABLE = True except ImportError: TF_AVAILABLE = False app = Flask(__name__) BALANCE = 10000.0 POSITIONS = [] ORDERS = [] SYMBOL_MAP = { "BTCUSDT": "BTC-USD", "ETHUSDT": "ETH-USD", "SOLUSDT": "SOL-USD", "ADAUSDT": "ADA-USD", "DOTUSDT": "DOT-USD", "AVAXUSDT": "AVAX-USD", "MATICUSDT": "MATIC-USD", "LINKUSDT": "LINK-USD", "XRPUSDT": "XRP-USD", "DOGEUSDT": "DOGE-USD" } @app.route("/", methods=["GET"]) def index(): return jsonify({"success": True, "message": "RapidLiveClient API running"}) @app.route("/health", methods=["GET"]) def health(): return jsonify({"success": True, "status": "healthy"}) @app.route("/api/balance", methods=["GET"]) def get_balance(): total = BALANCE for pos in POSITIONS: total += pos.get("pnl", 0) return jsonify({"success": True, "data": total}) @app.route("/api/positions", methods=["GET"]) def get_positions(): return jsonify({"success": True, "data": POSITIONS}) @app.route("/api/orders", methods=["GET"]) def get_orders(): return jsonify({"success": True, "data": ORDERS}) @app.route("/api/orders", methods=["POST"]) def create_order(): data = request.get_json() symbol = data.get("symbol", "BTCUSDT") side = data.get("side", "BUY") quantity = data.get("quantity", 0.01) order = { "id": f"ORD-{uuid.uuid4().hex[:8]}", "symbol": symbol, "side": side, "quantity": quantity, "status": "filled", "created_at": datetime.now().isoformat() } ORDERS.append(order) entry_price = 50000.0 position = { "id": f"POS-{uuid.uuid4().hex[:8]}", "symbol": symbol, "side": side, "quantity": quantity, "entry_price": entry_price, "current_price": entry_price, "pnl": 0.0, "opened_at": datetime.now().isoformat() } POSITIONS.append(position) return jsonify({"success": True, "data": order}) @app.route("/api/positions/close", methods=["POST"]) def close_position(): data = request.get_json() position_id = data.get("position_id") for pos in POSITIONS: if pos["id"] == position_id: pos["pnl"] = (pos["current_price"] - pos["entry_price"]) * pos["quantity"] POSITIONS.remove(pos) return jsonify({"success": True, "data": pos}) return jsonify({"success": False, "error": "Position not found"}) @app.route("/api/analyze", methods=["POST"]) def analyze_market(): data = request.get_json() symbol = data.get("symbol", "BTCUSDT") analysis = { "symbol": symbol, "recommendation": "BUY", "confidence": 0.75, "reason": "RSI oversold, trend bullish", "entry_price": 50000.0, "stop_loss": 47500.0, "take_profit": 55000.0, "risk_level": "MEDIUM" } return jsonify({"success": True, "data": analysis}) def scrape_crypto_news(symbol: str) -> list: """Web scraping de noticias relacionadas con la criptomoneda""" crypto_names = { "BTC": "bitcoin", "ETH": "ethereum", "SOL": "solana", "ADA": "cardano", "DOT": "polkadot", "AVAX": "avalanche", "MATIC": "polygon", "LINK": "chainlink", "XRP": "ripple", "DOGE": "dogecoin" } base_symbol = symbol.replace("USDT", "").replace("USD", "") crypto_name = crypto_names.get(base_symbol, base_symbol.lower()) news = [] sources = [ f"https://cryptonews.com/search/?q={crypto_name}", ] try: headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } response = requests.get( f"https://crypto.news/api/search/{crypto_name}/", headers=headers, timeout=5 ) if response.status_code == 200: data = response.json() for item in data.get("results", [])[:5]: news.append({ "title": item.get("title", ""), "source": item.get("source", ""), "url": item.get("url", "") }) except: pass if not news: news = [ {"title": f"Precio de {crypto_name.upper()} muestra volatilidad", "source": "Mercado", "url": ""}, {"title": f"Análisis técnico de {crypto_name.upper()} indica tendencia", "source": "Análisis", "url": ""}, {"title": f"Inversores observan {crypto_name.upper()} para próximos movimientos", "source": "Mercado", "url": ""} ] return news def prepare_lstm_data(data: np.ndarray, look_back: int = 60) -> tuple: """Prepara datos para LSTM""" X, y = [], [] for i in range(look_back, len(data)): X.append(data[i-look_back:i, 0]) y.append(data[i, 0]) return np.array(X), np.array(y) def build_lstm_model(look_back: int = 60) -> Sequential: """Construye modelo LSTM""" model = Sequential([ LSTM(50, return_sequences=True, input_shape=(look_back, 1)), Dropout(0.2), LSTM(50, return_sequences=False), Dropout(0.2), Dense(25), Dense(1) ]) model.compile(optimizer=Adam(learning_rate=0.001), loss='mean_squared_error') return model def predict_lstm(symbol: str, days: int = 30) -> dict: """Predicción LSTM para los próximos N días""" try: yf_symbol = SYMBOL_MAP.get(symbol, f"{symbol.replace('USDT', '')}-USD") ticker = yf.Ticker(yf_symbol) hist = ticker.history(period="2y") if len(hist) < 100: return {"success": False, "error": "Datos insuficientes"} close_prices = hist['Close'].values.reshape(-1, 1) close_prices = close_prices.astype('float32') look_back = min(60, len(close_prices) // 2) from sklearn.preprocessing import MinMaxScaler scaler = MinMaxScaler(feature_range=(0, 1)) scaled_data = scaler.fit_transform(close_prices) X, y = prepare_lstm_data(scaled_data, look_back) X = X.reshape(X.shape[0], X.shape[1], 1) model = build_lstm_model(look_back) try: model.fit(X, y, epochs=10, batch_size=32, verbose=0) except: model.fit(X, y, epochs=5, batch_size=32, verbose=0) last_60_days = scaled_data[-look_back:] predictions = [] for _ in range(days): X_pred = last_60_days.reshape(1, look_back, 1) pred = model.predict(X_pred, verbose=0)[0, 0] predictions.append(pred) last_60_days = np.append(last_60_days[1:], [[pred]], axis=0) predictions = scaler.inverse_transform(np.array(predictions).reshape(-1, 1)).flatten() current_price = float(close_prices[-1]) predicted_price = float(predictions[-1]) news = scrape_crypto_news(symbol) return { "success": True, "data": { "symbol": symbol, "current_price": current_price, "predicted_price": predicted_price, "price_change_pct": ((predicted_price - current_price) / current_price) * 100, "predictions": [ {"day": i+1, "price": float(p), "date": (datetime.now() + pd.Timedelta(days=i+1)).strftime("%Y-%m-%d")} for i, p in enumerate(predictions) ], "news": news, "model": "LSTM Deep Learning", "look_back": look_back, "training_data_points": len(close_prices) } } except Exception as e: return {"success": False, "error": str(e)} @app.route("/api/lstm-prediction", methods=["POST"]) def lstm_prediction(): """Endpoint para predicción LSTM""" if not TF_AVAILABLE: return jsonify({"success": False, "error": "TensorFlow no disponible"}) data = request.get_json() symbol = data.get("symbol", "BTCUSDT") days = data.get("days", 30) result = predict_lstm(symbol, days) return jsonify(result) if __name__ == "__main__": app.run(host="0.0.0.0", port=3000)