Tu Primer Indicador

mkdir cruce_ma
cd cruce_ma
pip install requests pandas

API_KEY = "tu_api_key_aqui"
BASE_URL = "https://api.innova-trading.com"

# Configuracion del indicador
SYMBOL = "EURUSD"
TIMEFRAME = 60  # 1 hora
MA_RAPIDA = 20
MA_LENTA = 50

import requests
import pandas as pd
from config import API_KEY, BASE_URL, SYMBOL, TIMEFRAME

def obtener_barras(limit=100):
    """Obtener barras OHLC de la API."""
    url = f"{BASE_URL}/api/external/bars/{SYMBOL}/{TIMEFRAME}"
    headers = {"Authorization": f"Bearer {API_KEY}"}

    response = requests.get(url, params={"limit": limit}, headers=headers)
    response.raise_for_status()

    data = response.json()
    df = pd.DataFrame(data["bars"])
    return df

# Probarlo
if __name__ == "__main__":
    barras = obtener_barras(100)
    print(f"Obtenidas {len(barras)} barras")
    print(barras.tail())

from config import MA_RAPIDA, MA_LENTA

def calcular_sma(df, periodo, columna="close"):
    """Calcular Media Movil Simple."""
    return df[columna].rolling(window=periodo).mean()

def agregar_medias_moviles(df):
    """Agregar MAs rapida y lenta al dataframe."""
    df["ma_rapida"] = calcular_sma(df, MA_RAPIDA)
    df["ma_lenta"] = calcular_sma(df, MA_LENTA)
    return df

# Probarlo
if __name__ == "__main__":
    barras = obtener_barras(100)
    barras = agregar_medias_moviles(barras)
    print(barras[["time", "close", "ma_rapida", "ma_lenta"]].tail())

def detectar_cruces(df):
    """
    Detectar cruces de MA y generar puntos de senal.

    Retorna lista de puntos de senal en formato API.
    """
    puntos = []

    for i in range(1, len(df)):
        prev = df.iloc[i - 1]
        curr = df.iloc[i]

        # Saltar si las MAs no estan calculadas aun
        if pd.isna(prev["ma_rapida"]) or pd.isna(curr["ma_lenta"]):
            continue

        # Cruce alcista: MA rapida cruza por encima de MA lenta
        diff_prev = prev["ma_rapida"] - prev["ma_lenta"]
        diff_curr = curr["ma_rapida"] - curr["ma_lenta"]

        if diff_prev <= 0 and diff_curr > 0:
            # Senal de COMPRA
            puntos.append({
                "time": int(curr["time"]),
                "type": "low",
                "price": float(curr["low"]) - 0.0005,
                "label": "COMPRA",
                "color": "#3b82f6",
                "shape": "arrowUp",
                "size": 2
            })

        elif diff_prev >= 0 and diff_curr < 0:
            # Senal de VENTA
            puntos.append({
                "time": int(curr["time"]),
                "type": "high",
                "price": float(curr["high"]) + 0.0005,
                "label": "VENTA",
                "color": "#f97316",
                "shape": "arrowDown",
                "size": 2
            })

    return puntos

def enviar_indicador(puntos):
    """Enviar puntos de indicador a la API."""
    if not puntos:
        print("No hay senales para enviar")
        return None

    url = f"{BASE_URL}/api/external/indicators/cruce_ma"
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }

    payload = {
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "indicator_name": f"Cruce MA ({MA_RAPIDA}/{MA_LENTA})",
        "version": "1.0",
        "points": puntos,
        "metadata": {
            "ma_rapida": MA_RAPIDA,
            "ma_lenta": MA_LENTA,
            "total_senales": len(puntos)
        }
    }

    response = requests.post(url, json=payload, headers=headers)
    response.raise_for_status()
    return response.json()

def main():
    print("=" * 50)
    print("Indicador Cruce de Medias Moviles")
    print("=" * 50)

    # 1. Obtener datos
    print("\n1. Obteniendo datos de mercado...")
    barras = obtener_barras(500)
    print(f"   Obtenidas {len(barras)} barras")

    # 2. Calcular MAs
    print("\n2. Calculando medias moviles...")
    barras = agregar_medias_moviles(barras)

    # 3. Detectar cruces
    print("\n3. Detectando cruces...")
    senales = detectar_cruces(barras)
    print(f"   Encontradas {len(senales)} senales")

    if senales:
        print("\n   Senales recientes:")
        for s in senales[-5:]:
            print(f"   - {s['label']} a {s['price']:.5f}")

    # 4. Enviar a API
    print("\n4. Enviando a InnovaTrading...")
    resultado = enviar_indicador(senales)

    if resultado:
        print(f"   Exito! {resultado['points_received']} puntos enviados")
        print(f"   Expira: {resultado['expires_at']}")
    else:
        print("   No hay senales para enviar")

    print("\n" + "=" * 50)
    print("Listo! Revisa tu chart para ver las senales.")
    print("=" * 50)

if __name__ == "__main__":
    main()