Your First Indicator

mkdir ma_crossover
cd ma_crossover
pip install requests pandas

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

# Indicator settings
SYMBOL = "EURUSD"
TIMEFRAME = 60  # 1 hour
FAST_MA = 20
SLOW_MA = 50

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

def get_bars(limit=100):
    """Fetch OHLC bars from the 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

# Test it
if __name__ == "__main__":
    bars = get_bars(100)
    print(f"Fetched {len(bars)} bars")
    print(bars.tail())

from config import FAST_MA, SLOW_MA

def calculate_sma(df, period, column="close"):
    """Calculate Simple Moving Average."""
    return df[column].rolling(window=period).mean()

def add_moving_averages(df):
    """Add fast and slow MAs to dataframe."""
    df["fast_ma"] = calculate_sma(df, FAST_MA)
    df["slow_ma"] = calculate_sma(df, SLOW_MA)
    return df

# Test it
if __name__ == "__main__":
    bars = get_bars(100)
    bars = add_moving_averages(bars)
    print(bars[["time", "close", "fast_ma", "slow_ma"]].tail())

def detect_crossovers(df):
    """
    Detect MA crossovers and generate signal points.

    Returns list of signal points in API format.
    """
    points = []

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

        # Skip if MAs are not yet calculated
        if pd.isna(prev["fast_ma"]) or pd.isna(curr["slow_ma"]):
            continue

        # Bullish crossover: fast MA crosses above slow MA
        prev_diff = prev["fast_ma"] - prev["slow_ma"]
        curr_diff = curr["fast_ma"] - curr["slow_ma"]

        if prev_diff <= 0 and curr_diff > 0:
            # BUY signal
            points.append({
                "time": int(curr["time"]),
                "type": "low",
                "price": float(curr["low"]) - 0.0005,  # Slightly below candle
                "label": "BUY",
                "color": "#3b82f6",
                "shape": "arrowUp",
                "size": 2
            })

        elif prev_diff >= 0 and curr_diff < 0:
            # SELL signal
            points.append({
                "time": int(curr["time"]),
                "type": "high",
                "price": float(curr["high"]) + 0.0005,  # Slightly above candle
                "label": "SELL",
                "color": "#f97316",
                "shape": "arrowDown",
                "size": 2
            })

    return points

def submit_indicator(points):
    """Submit indicator points to the API."""
    if not points:
        print("No signals to submit")
        return None

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

    payload = {
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "indicator_name": f"MA Crossover ({FAST_MA}/{SLOW_MA})",
        "version": "1.0",
        "points": points,
        "metadata": {
            "fast_period": FAST_MA,
            "slow_period": SLOW_MA,
            "total_signals": len(points)
        }
    }

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

def main():
    print("=" * 50)
    print("MA Crossover Indicator")
    print("=" * 50)

    # 1. Fetch data
    print("\n1. Fetching market data...")
    bars = get_bars(500)
    print(f"   Fetched {len(bars)} bars")

    # 2. Calculate MAs
    print("\n2. Calculating moving averages...")
    bars = add_moving_averages(bars)

    # 3. Detect crossovers
    print("\n3. Detecting crossovers...")
    signals = detect_crossovers(bars)
    print(f"   Found {len(signals)} signals")

    if signals:
        print("\n   Recent signals:")
        for s in signals[-5:]:
            print(f"   - {s['label']} at {s['price']:.5f}")

    # 4. Submit to API
    print("\n4. Submitting to InnovaTrading...")
    result = submit_indicator(signals)

    if result:
        print(f"   Success! {result['points_received']} points submitted")
        print(f"   Expires at: {result['expires_at']}")
    else:
        print("   No signals to submit")

    print("\n" + "=" * 50)
    print("Done! Check your chart to see the signals.")
    print("=" * 50)

if __name__ == "__main__":
    main()