How to Build a Crypto Trading Bot with CoinGlass API

Why not just use Binance API, OKX API, Bybit API, or another exchange API?

Exchange API = execution layer
CoinGlass API = market intelligence layer

Can I place this order?
What is my account balance?
What is my current position?

Should I place this order?
Is the market crowded?
Is leverage building up?
Are traders being liquidated?
Is this breakout healthy?

Data Layer
    ↓
Feature Layer
    ↓
Signal Layer
    ↓
Risk Layer
    ↓
Execution Layer
    ↓
Logging / Monitoring Layer

CoinGlass API
  - Open Interest
  - Funding Rate
  - Liquidation Data
  - Long/Short Ratio
  - Order Flow

Exchange API
  - Price Candles
  - Account Balance
  - Order Placement
  - Position Management

Trading Bot
  - Feature Engineering
  - Signal Generation
  - Risk Filtering
  - Position Sizing
  - Execution

If price breaks above moving average → BUY
If price breaks below moving average → SELL

If price gives BUY signal
AND funding rate is not extremely high
AND open interest supports the move
AND liquidation risk is not extreme
THEN allow the trade

Otherwise, reduce size or skip the trade

Did price go up?

Is this price move healthy?
Is leverage supporting it?
Is one side too crowded?
Is the market entering a liquidation cascade?

Is new leverage entering the market?

BTC price rises 3%.
Open interest also rises.
Funding rate remains neutral.

BTC price rises 3%.
Open interest rises sharply.
Funding rate becomes extremely positive.
Long/short ratio becomes heavily long.

Funding high = short
Funding low = long

Funding high = long side may be crowded
Funding low = short side may be crowded

Which side is being forced out?

https://open-api-v4.coinglass.com

pip install requests pandas python-dotenv

COINGLASS_API_KEY=your_api_key_here

bot.py

import os
import requests
import pandas as pd
from dotenv import load_dotenv

load_dotenv()

BASE_URL = "https://open-api-v4.coinglass.com"
API_KEY = os.getenv("COINGLASS_API_KEY")

if not API_KEY:
    raise RuntimeError("Missing COINGLASS_API_KEY environment variable")

HEADERS = {
    "CG-API-KEY": API_KEY,
    "Accept": "application/json"
}


def request_coinglass(endpoint, params=None):
    """
    Send a GET request to CoinGlass API.
    """
    url = f"{BASE_URL}{endpoint}"

    response = requests.get(
        url,
        headers=HEADERS,
        params=params,
        timeout=10
    )

    response.raise_for_status()
    return response.json()

import time


def safe_request_coinglass(endpoint, params=None, retries=3, sleep_seconds=2):
    """
    Send a GET request with retry logic.
    """
    last_error = None

    for attempt in range(retries):
        try:
            return request_coinglass(endpoint, params)

        except requests.RequestException as error:
            last_error = error
            print(f"Request failed: attempt {attempt + 1}/{retries}: {error}")
            time.sleep(sleep_seconds)

    raise last_error

/api/futures/openInterest/ohlc-history
/api/futures/openInterest/aggregated-history
/api/futures/openInterest/exchange-list

def fetch_open_interest(symbol="BTC", interval="1h", limit=100):
    endpoint = "/api/futures/openInterest/ohlc-history"

    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }

    return safe_request_coinglass(endpoint, params)

def to_dataframe(raw):
    """
    Convert a CoinGlass API response to a pandas DataFrame.
    Adjust this function based on the actual response shape.
    """
    data = raw.get("data", [])

    if isinstance(data, dict):
        # Some APIs may return nested structures.
        # Adjust based on actual response.
        rows = data.get("list", [])
    else:
        rows = data

    df = pd.DataFrame(rows)

    if df.empty:
        return df

    if "time" in df.columns:
        df["time"] = pd.to_datetime(df["time"], unit="ms", errors="coerce")

    return df

def add_open_interest_features(df):
    data = df.copy()

    # Adjust field names based on actual API response.
    if "close" in data.columns:
        data["oi_close"] = pd.to_numeric(data["close"], errors="coerce")
    elif "openInterest" in data.columns:
        data["oi_close"] = pd.to_numeric(data["openInterest"], errors="coerce")
    else:
        raise ValueError("No recognizable open interest field found")

    data["oi_change"] = data["oi_close"].pct_change()
    data["oi_change_24"] = data["oi_close"].pct_change(24)

    return data

/api/futures/fundingRate/ohlc-history
/api/futures/fundingRate/oi-weight-ohlc-history
/api/futures/fundingRate/exchange-list

def fetch_funding_rate(symbol="BTC", interval="1h", limit=100):
    endpoint = "/api/futures/fundingRate/oi-weight-ohlc-history"

    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }

    return safe_request_coinglass(endpoint, params)

def zscore(series, window=24):
    mean = series.rolling(window).mean()
    std = series.rolling(window).std()
    return (series - mean) / std


def add_funding_features(df, window=24):
    data = df.copy()

    # Adjust field name based on actual API response.
    if "close" in data.columns:
        data["funding_close"] = pd.to_numeric(data["close"], errors="coerce")
    elif "fundingRate" in data.columns:
        data["funding_close"] = pd.to_numeric(data["fundingRate"], errors="coerce")
    else:
        raise ValueError("No recognizable funding rate field found")

    data["funding_z"] = zscore(data["funding_close"], window)

    return data

/api/futures/liquidation/history
/api/futures/liquidation/aggregated-history
/api/futures/liquidation/heatmap/model2

def fetch_liquidation_history(
    symbol="BTC",
    exchanges="Binance,OKX,Bybit",
    interval="1h",
    limit=100
):
    endpoint = "/api/futures/liquidation/aggregated-history"

    params = {
        "exchange_list": exchanges,
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }

    return safe_request_coinglass(endpoint, params)

def add_liquidation_features(df, window=24):
    data = df.copy()

    rename_map = {
        "longLiquidation": "long_liquidation",
        "shortLiquidation": "short_liquidation",
        "long_liq": "long_liquidation",
        "short_liq": "short_liquidation"
    }

    data = data.rename(columns=rename_map)

    if "long_liquidation" not in data.columns:
        raise ValueError("No long liquidation field found")

    if "short_liquidation" not in data.columns:
        raise ValueError("No short liquidation field found")

    data["long_liquidation"] = pd.to_numeric(
        data["long_liquidation"],
        errors="coerce"
    )

    data["short_liquidation"] = pd.to_numeric(
        data["short_liquidation"],
        errors="coerce"
    )

    data["long_liq_z"] = zscore(data["long_liquidation"], window)
    data["short_liq_z"] = zscore(data["short_liquidation"], window)

    data["total_liquidation"] = (
        data["long_liquidation"] + data["short_liquidation"]
    )

    data["total_liq_z"] = zscore(data["total_liquidation"], window)

    return data

def prepare_time_column(df):
    data = df.copy()

    if "time" not in data.columns:
        raise ValueError("Missing time column")

    data = data.dropna(subset=["time"])
    data = data.sort_values("time")

    return data


def merge_market_data(oi_df, funding_df, liquidation_df):
    oi = prepare_time_column(oi_df)
    funding = prepare_time_column(funding_df)
    liquidation = prepare_time_column(liquidation_df)

    merged = pd.merge_asof(
        oi,
        funding,
        on="time",
        direction="nearest",
        tolerance=pd.Timedelta("10min"),
        suffixes=("_oi", "_funding")
    )

    merged = pd.merge_asof(
        merged,
        liquidation,
        on="time",
        direction="nearest",
        tolerance=pd.Timedelta("10min")
    )

    return merged

def generate_price_signal(price_df):
    """
    Simple moving average signal.
    price_df must contain a 'close' column.
    """
    data = price_df.copy()

    data["close"] = pd.to_numeric(data["close"], errors="coerce")
    data["ma_fast"] = data["close"].rolling(20).mean()
    data["ma_slow"] = data["close"].rolling(60).mean()

    latest = data.iloc[-1]

    if latest["ma_fast"] > latest["ma_slow"]:
        return "BUY"

    if latest["ma_fast"] < latest["ma_slow"]:
        return "SELL"

    return "HOLD"

def classify_derivatives_state(row):
    funding_z = row.get("funding_z", 0)
    long_liq_z = row.get("long_liq_z", 0)
    short_liq_z = row.get("short_liq_z", 0)
    total_liq_z = row.get("total_liq_z", 0)
    oi_change = row.get("oi_change", 0)

    if pd.isna(funding_z):
        funding_z = 0
    if pd.isna(long_liq_z):
        long_liq_z = 0
    if pd.isna(short_liq_z):
        short_liq_z = 0
    if pd.isna(total_liq_z):
        total_liq_z = 0
    if pd.isna(oi_change):
        oi_change = 0

    if total_liq_z > 3:
        return "RISK_OFF"

    if funding_z > 2 and oi_change > 0:
        return "LONG_CROWDED"

    if funding_z < -2 and oi_change > 0:
        return "SHORT_CROWDED"

    if short_liq_z > 2 and funding_z < 0:
        return "SHORT_SQUEEZE"

    if long_liq_z > 2 and funding_z > 0:
        return "LONG_SQUEEZE"

    return "NEUTRAL"

def final_trading_decision(price_signal, derivatives_state):
    """
    Combine price signal with derivatives market state.
    """

    if derivatives_state == "RISK_OFF":
        return "HOLD"

    if price_signal == "BUY" and derivatives_state == "LONG_CROWDED":
        return "HOLD"

    if price_signal == "SELL" and derivatives_state == "SHORT_CROWDED":
        return "HOLD"

    if price_signal == "BUY" and derivatives_state == "SHORT_SQUEEZE":
        return "BUY"

    if price_signal == "SELL" and derivatives_state == "LONG_SQUEEZE":
        return "SELL"

    if derivatives_state in ["NEUTRAL", "SHORT_SQUEEZE", "LONG_SQUEEZE"]:
        return price_signal

    return "HOLD"

def position_size_multiplier(derivatives_state):
    multipliers = {
        "NEUTRAL": 1.0,
        "SHORT_SQUEEZE": 1.0,
        "LONG_SQUEEZE": 1.0,
        "LONG_CROWDED": 0.4,
        "SHORT_CROWDED": 0.4,
        "RISK_OFF": 0.0
    }

    return multipliers.get(derivatives_state, 0.5)

base_position_size = 1000  # USDT notional
derivatives_state = "LONG_CROWDED"

final_size = base_position_size * position_size_multiplier(derivatives_state)

print(final_size)

import os
import time
import requests
import pandas as pd
from dotenv import load_dotenv

load_dotenv()

BASE_URL = "https://open-api-v4.coinglass.com"
API_KEY = os.getenv("COINGLASS_API_KEY")

if not API_KEY:
    raise RuntimeError("Missing COINGLASS_API_KEY environment variable")

HEADERS = {
    "CG-API-KEY": API_KEY,
    "Accept": "application/json"
}


def request_coinglass(endpoint, params=None):
    url = f"{BASE_URL}{endpoint}"

    response = requests.get(
        url,
        headers=HEADERS,
        params=params,
        timeout=10
    )

    response.raise_for_status()
    return response.json()


def safe_request_coinglass(endpoint, params=None, retries=3, sleep_seconds=2):
    last_error = None

    for attempt in range(retries):
        try:
            return request_coinglass(endpoint, params)

        except requests.RequestException as error:
            last_error = error
            print(f"Request failed: attempt {attempt + 1}/{retries}: {error}")
            time.sleep(sleep_seconds)

    raise last_error


def to_dataframe(raw):
    data = raw.get("data", [])

    if isinstance(data, dict):
        rows = data.get("list", [])
    else:
        rows = data

    df = pd.DataFrame(rows)

    if df.empty:
        return df

    if "time" in df.columns:
        df["time"] = pd.to_datetime(df["time"], unit="ms", errors="coerce")

    return df


def zscore(series, window=24):
    mean = series.rolling(window).mean()
    std = series.rolling(window).std()
    return (series - mean) / std


def fetch_open_interest(symbol="BTC", interval="1h", limit=100):
    endpoint = "/api/futures/openInterest/ohlc-history"

    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }

    return safe_request_coinglass(endpoint, params)


def fetch_funding_rate(symbol="BTC", interval="1h", limit=100):
    endpoint = "/api/futures/fundingRate/oi-weight-ohlc-history"

    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }

    return safe_request_coinglass(endpoint, params)


def fetch_liquidation_history(
    symbol="BTC",
    exchanges="Binance,OKX,Bybit",
    interval="1h",
    limit=100
):
    endpoint = "/api/futures/liquidation/aggregated-history"

    params = {
        "exchange_list": exchanges,
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }

    return safe_request_coinglass(endpoint, params)


def add_open_interest_features(df):
    data = df.copy()

    if "close" in data.columns:
        data["oi_close"] = pd.to_numeric(data["close"], errors="coerce")
    elif "openInterest" in data.columns:
        data["oi_close"] = pd.to_numeric(data["openInterest"], errors="coerce")
    else:
        raise ValueError("No recognizable open interest field found")

    data["oi_change"] = data["oi_close"].pct_change()
    data["oi_change_24"] = data["oi_close"].pct_change(24)

    return data


def add_funding_features(df, window=24):
    data = df.copy()

    if "close" in data.columns:
        data["funding_close"] = pd.to_numeric(data["close"], errors="coerce")
    elif "fundingRate" in data.columns:
        data["funding_close"] = pd.to_numeric(
            data["fundingRate"],
            errors="coerce"
        )
    else:
        raise ValueError("No recognizable funding rate field found")

    data["funding_z"] = zscore(data["funding_close"], window)

    return data


def add_liquidation_features(df, window=24):
    data = df.copy()

    data = data.rename(columns={
        "longLiquidation": "long_liquidation",
        "shortLiquidation": "short_liquidation",
        "long_liq": "long_liquidation",
        "short_liq": "short_liquidation"
    })

    if "long_liquidation" not in data.columns:
        raise ValueError("No long liquidation field found")

    if "short_liquidation" not in data.columns:
        raise ValueError("No short liquidation field found")

    data["long_liquidation"] = pd.to_numeric(
        data["long_liquidation"],
        errors="coerce"
    )

    data["short_liquidation"] = pd.to_numeric(
        data["short_liquidation"],
        errors="coerce"
    )

    data["long_liq_z"] = zscore(data["long_liquidation"], window)
    data["short_liq_z"] = zscore(data["short_liquidation"], window)

    data["total_liquidation"] = (
        data["long_liquidation"] + data["short_liquidation"]
    )

    data["total_liq_z"] = zscore(data["total_liquidation"], window)

    return data


def prepare_time_column(df):
    data = df.copy()

    if "time" not in data.columns:
        raise ValueError("Missing time column")

    data = data.dropna(subset=["time"])
    data = data.sort_values("time")

    return data


def merge_market_data(oi_df, funding_df, liquidation_df):
    oi = prepare_time_column(oi_df)
    funding = prepare_time_column(funding_df)
    liquidation = prepare_time_column(liquidation_df)

    merged = pd.merge_asof(
        oi,
        funding,
        on="time",
        direction="nearest",
        tolerance=pd.Timedelta("10min"),
        suffixes=("_oi", "_funding")
    )

    merged = pd.merge_asof(
        merged,
        liquidation,
        on="time",
        direction="nearest",
        tolerance=pd.Timedelta("10min")
    )

    return merged


def classify_derivatives_state(row):
    funding_z = row.get("funding_z", 0)
    long_liq_z = row.get("long_liq_z", 0)
    short_liq_z = row.get("short_liq_z", 0)
    total_liq_z = row.get("total_liq_z", 0)
    oi_change = row.get("oi_change", 0)

    values = [funding_z, long_liq_z, short_liq_z, total_liq_z, oi_change]
    values = [0 if pd.isna(v) else v for v in values]

    funding_z, long_liq_z, short_liq_z, total_liq_z, oi_change = values

    if total_liq_z > 3:
        return "RISK_OFF"

    if funding_z > 2 and oi_change > 0:
        return "LONG_CROWDED"

    if funding_z < -2 and oi_change > 0:
        return "SHORT_CROWDED"

    if short_liq_z > 2 and funding_z < 0:
        return "SHORT_SQUEEZE"

    if long_liq_z > 2 and funding_z > 0:
        return "LONG_SQUEEZE"

    return "NEUTRAL"


def final_trading_decision(price_signal, derivatives_state):
    if derivatives_state == "RISK_OFF":
        return "HOLD"

    if price_signal == "BUY" and derivatives_state == "LONG_CROWDED":
        return "HOLD"

    if price_signal == "SELL" and derivatives_state == "SHORT_CROWDED":
        return "HOLD"

    if price_signal == "BUY" and derivatives_state == "SHORT_SQUEEZE":
        return "BUY"

    if price_signal == "SELL" and derivatives_state == "LONG_SQUEEZE":
        return "SELL"

    if derivatives_state in ["NEUTRAL", "SHORT_SQUEEZE", "LONG_SQUEEZE"]:
        return price_signal

    return "HOLD"


def position_size_multiplier(derivatives_state):
    multipliers = {
        "NEUTRAL": 1.0,
        "SHORT_SQUEEZE": 1.0,
        "LONG_SQUEEZE": 1.0,
        "LONG_CROWDED": 0.4,
        "SHORT_CROWDED": 0.4,
        "RISK_OFF": 0.0
    }

    return multipliers.get(derivatives_state, 0.5)


def run_bot_once(symbol="BTC"):
    oi_raw = fetch_open_interest(symbol=symbol, interval="1h", limit=100)
    funding_raw = fetch_funding_rate(symbol=symbol, interval="1h", limit=100)
    liquidation_raw = fetch_liquidation_history(
        symbol=symbol,
        interval="1h",
        limit=100
    )

    oi_df = add_open_interest_features(to_dataframe(oi_raw))
    funding_df = add_funding_features(to_dataframe(funding_raw))
    liquidation_df = add_liquidation_features(to_dataframe(liquidation_raw))

    market_df = merge_market_data(oi_df, funding_df, liquidation_df)

    if market_df.empty:
        raise ValueError("Merged market data is empty")

    latest = market_df.iloc[-1]

    # Example only. Replace this with your real price signal.
    price_signal = "BUY"

    derivatives_state = classify_derivatives_state(latest)
    final_decision = final_trading_decision(price_signal, derivatives_state)

    base_position_size = 1000
    multiplier = position_size_multiplier(derivatives_state)
    final_position_size = base_position_size * multiplier

    result = {
        "symbol": symbol,
        "time": latest["time"],
        "price_signal": price_signal,
        "derivatives_state": derivatives_state,
        "final_decision": final_decision,
        "position_size": final_position_size
    }

    return result


if __name__ == "__main__":
    result = run_bot_once("BTC")

    print("Bot result:")
    for key, value in result.items():
        print(f"{key}: {value}")

def execute_trade(decision, symbol, position_size):
    """
    Placeholder execution function.
    Replace with your exchange API logic.
    """

    if decision == "BUY":
        print(f"Place BUY order for {symbol}, size={position_size}")

    elif decision == "SELL":
        print(f"Place SELL order for {symbol}, size={position_size}")

    else:
        print(f"No trade for {symbol}")

result = run_bot_once("BTC")

execute_trade(
    decision=result["final_decision"],
    symbol=result["symbol"],
    position_size=result["position_size"]
)

def risk_filter(account_state, market_state):
    """
    Example risk control logic.
    """

    if account_state["daily_loss_pct"] < -3:
        return "STOP_TRADING"

    if account_state["open_positions"] >= account_state["max_positions"]:
        return "NO_NEW_POSITIONS"

    if market_state["derivatives_state"] == "RISK_OFF":
        return "NO_NEW_POSITIONS"

    return "ALLOW"

Signal decides what you want to do.
Risk decides whether you are allowed to do it.

def backtest(market_df, price_signals):
    equity = 10000
    position = 0
    trades = []

    for i in range(len(market_df)):
        row = market_df.iloc[i]
        price_signal = price_signals[i]

        derivatives_state = classify_derivatives_state(row)
        decision = final_trading_decision(price_signal, derivatives_state)
        size_multiplier = position_size_multiplier(derivatives_state)

        trades.append({
            "time": row["time"],
            "price_signal": price_signal,
            "derivatives_state": derivatives_state,
            "decision": decision,
            "size_multiplier": size_multiplier
        })

    return pd.DataFrame(trades)

def check_data_freshness(latest_time, max_age_minutes=90):
    now = pd.Timestamp.utcnow()

    if latest_time.tzinfo is None:
        latest_time = latest_time.tz_localize("UTC")

    age = now - latest_time

    if age > pd.Timedelta(minutes=max_age_minutes):
        raise ValueError(f"Data is stale: latest data age is {age}")

    return True

Funding high → short
Funding low → long

Funding high → long side may be crowded
Funding low → short side may be crowded
Confirm with OI, price, and liquidation data

Price signal
+ OI confirmation
+ Funding filter
+ Liquidation risk filter
+ Position sizing rule

Stop trading if API data is stale.
Stop trading if daily loss exceeds threshold.
Stop trading if liquidation risk is extreme.
Stop trading if exchange order API behaves unexpectedly.

Trade only when price trend is supported by derivatives data.

Detect possible short squeeze setups.

Avoid long exposure when the long side is overcrowded.

Detect cross-exchange funding rate opportunities.

Pause trading during extreme derivatives stress.

If long/short ratio is extremely long
AND funding is positive
AND OI is rising
THEN reduce long exposure.

If price breaks resistance
AND taker buy volume rises
AND short liquidations rise
THEN confirm short squeeze.

If price approaches large liquidation cluster
THEN reduce leverage or tighten stop.

If order book depth is thin
THEN reduce order size or use limit orders.

If ETF inflows are strong
AND derivatives data is not overheated
THEN increase bullish confidence.

Price is above the moving average, so buy.

Price is above the moving average.
Open interest is rising.
Funding rate is neutral.
Liquidations are normal.
Market structure is healthy.
Buy signal is allowed.

Price is above the moving average.
But funding is extremely positive.
Open interest is rising too fast.
Long side is crowded.
Do not chase.