用资金费率 + 清算数据让你的交易机器人胜率提升：CoinGlass API 实战接入

m0_38016714

524人浏览 · 2026-05-25 14:45:40

m0_38016714 · 2026-05-25 14:45:40 发布

在加密货币市场里，很多交易机器人只盯着价格、成交量、均线、RSI、MACD 等传统指标。但如果你的交易机器人主要交易 BTC、ETH 或主流山寨币合约，只看现货价格是不够的。

原因很简单：加密市场的价格波动，很大一部分来自衍生品市场的杠杆行为。

当市场多头过度拥挤时，价格只要小幅回调，就可能触发大规模多头清算；当市场空头过度拥挤时，价格向上突破，也可能引发空头踩踏。对交易机器人来说，真正重要的不是“价格已经涨了多少”，而是：

当前市场多头是否过热？
空头是否被挤压？
杠杆资金是否正在聚集？
某个方向是否存在被清算的风险？
当前信号是趋势延续，还是反向陷阱？

这就是为什么 资金费率 Funding Rate 和 清算数据 Liquidation Data 对交易机器人非常重要。

通过 CoinGlass API，开发者可以获取合约市场中的资金费率、清算历史、清算热力图、持仓量、多空比、订单簿等衍生品数据。CoinGlass 官方资料显示，其 API 覆盖合约、现货、期权、ETF 和链上市场，并提供 L2/L3 订单簿、清算热力图、足迹图、流动性热力图等高阶指标，适用于量化交易、研究分析、数据建模和风险管理。(coinglass)

本文会重点讲：

为什么交易机器人需要资金费率和清算数据
资金费率如何辅助判断市场情绪
清算数据如何识别短线风险和反转机会
如何用 CoinGlass API 接入相关数据
如何把这些数据转化为交易机器人可用的信号
Python 示例代码和策略框架

注意：本文中的“胜率提升”不是承诺收益，也不是投资建议。更准确地说，资金费率和清算数据可以帮助交易机器人减少盲目入场、识别拥挤交易和优化风控，从而提高策略质量。

一、为什么交易机器人不能只看价格？

很多基础交易机器人采用的是价格驱动逻辑，例如：

类型	常见信号	问题
趋势策略	均线金叉、突破前高	容易追在杠杆过热区
震荡策略	RSI 超买超卖	遇到强趋势容易逆势亏损
网格策略	固定区间买低卖高	遇到单边行情容易被打穿
高频策略	短线盘口变化	忽略全市场杠杆风险
套利策略	价差、费率差	需要更完整的市场数据支持

价格本身只是结果，背后的推动因素往往来自：

合约市场杠杆
多空双方成本
强制平仓
资金费率变化
订单簿流动性
大户持仓变化

如果一个交易机器人只看 K 线，那么它看到的是“已经发生的价格变化”；如果它同时接入衍生品数据，它就有机会看到价格背后的“压力来源”。

举个例子：

当 BTC 连续上涨，价格突破前高，普通趋势机器人可能会继续追多。但如果此时资金费率已经明显偏高，且清算热力图显示上方流动性已经被扫，下方存在大量多头清算区域，那么追多风险会明显增加。

相反，如果 BTC 快速下跌后出现大量多头清算，但价格没有继续破位，资金费率从极端正值回落到中性区间，这反而可能说明杠杆已经被清洗，市场进入新的平衡阶段。

这类判断，单靠价格很难完成。

二、资金费率是什么？为什么适合交易机器人？

资金费率是永续合约市场中多空双方定期支付的费用。简单理解：

资金费率状态	市场含义	可能解读
资金费率为正	多头支付空头	市场偏多，多头更拥挤
资金费率为负	空头支付多头	市场偏空，空头更拥挤
资金费率极高	多头成本升高	可能出现多头拥挤风险
资金费率极低或负值	空头成本升高	可能出现空头挤压风险
资金费率回归中性	杠杆情绪降温	市场可能进入新平衡

资金费率不是单纯的做多或做空信号，而是一个 市场拥挤度指标。

交易机器人可以用它来判断：

当前市场是否过度看多
当前市场是否过度看空
趋势行情是否健康
是否存在反向挤压风险
是否应该降低仓位
是否过滤掉高风险信号

CoinGlass 官方 API 文档中，资金费率模块包含 Funding Rate History、OI Weight History、Vol Weight History、Exchange List、Arbitrage 等接口分类。(GitHub) 这意味着开发者不仅可以看单一交易所资金费率，也可以结合持仓量加权、成交量加权等方式观察更综合的市场情绪。

三、清算数据是什么？为什么它比普通成交量更重要？

清算数据指的是合约市场中因保证金不足而被强制平仓的数据。它和普通成交量不同。

普通成交量只告诉你“市场发生了交易”，但清算数据告诉你：

哪个方向的杠杆被打爆了
市场是否发生了强制踩踏
多头或空头是否已经被清洗
某个价格区域是否存在集中爆仓风险
极端波动是否来自真实买卖，还是被动清算

例如：

市场现象	清算数据解读
价格急跌 + 多头大量清算	多头杠杆被清洗
价格急涨 + 空头大量清算	空头被挤压
价格横盘 + 清算减少	杠杆压力下降
价格小幅波动 + 清算异常放大	市场杠杆过高
清算后价格快速收回	可能是假突破或流动性扫单

CoinGlass 文档显示，Coin Liquidation History 接口用于提供多个交易所中某个币种的多空清算聚合历史数据。该接口路径为 /api/futures/liquidation/aggregated-history，支持 exchange_list、symbol、interval、limit、start_time、end_time 等参数。(CoinGlass-API)

这类数据非常适合交易机器人做风控过滤。

例如：

如果 15 分钟内多头清算突然放大，机器人可以暂停追空，避免追在清算尾部。
如果空头清算持续增加，且价格突破关键阻力，机器人可以判断可能出现空头挤压。
如果资金费率极高，同时多头清算开始放大，机器人可以减少多头仓位或提高止损敏感度。

四、资金费率 + 清算数据如何组合成交易信号？

单独使用资金费率，容易误判。

因为资金费率高，不代表价格马上下跌；资金费率低，也不代表价格马上上涨。强趋势行情中，资金费率可能长时间维持高位。

单独使用清算数据，也容易滞后。

因为清算通常发生在价格剧烈波动之后。如果只在清算发生后入场，可能已经错过最好的位置。

所以更好的方式是：把资金费率作为市场情绪背景，把清算数据作为风险触发器。

1. 多头拥挤风险模型

条件	含义
资金费率持续为正且升高	市场多头拥挤
价格连续上涨	多头情绪强化
持仓量同步上升	杠杆资金进入
多头清算开始放大	多头被动平仓开始
价格跌破短周期支撑	风险释放触发

机器人动作：

场景	动作
已持有多单	减仓或收紧止损
准备开多	暂停入场
有做空策略	等待确认后小仓位试空
网格策略	缩小仓位，扩大网格间距

2. 空头挤压模型

条件	含义
资金费率为负	市场偏空
空头清算开始增加	空头被迫平仓
价格突破阻力	可能触发 short squeeze
成交量放大	趋势确认
资金费率从负值回升	空头情绪修复

机器人动作：

场景	动作
趋势机器人	顺势跟随突破
套利机器人	监控费率回归
空头策略	降低杠杆或退出
风控系统	提醒空头挤压风险

3. 杠杆清洗后反转模型

条件	含义
某方向出现大量清算	杠杆被清洗
价格没有继续破位	卖压或买压衰竭
资金费率回归中性	市场情绪降温
持仓量下降	杠杆出清
价格重新站回关键位置	反转可能增强

机器人动作：

场景	动作
趋势策略	等待重新确认方向
均值回归策略	小仓位试探反弹
风控系统	降低极端行情权重
高频策略	关注盘口恢复情况

五、CoinGlass API 可以提供哪些关键数据？

根据 CoinGlass 官方 API 资料和公开文档，衍生品相关数据覆盖持仓量、资金费率、多空比、清算、订单簿、主动买卖、CVD 等多个模块。官方 GitHub 文档也列出了 Futures - Funding Rate、Futures - Liquidation、Futures - Open Interest、Futures - Order Book、Futures - Taker Buy/Sell 等接口分类。(GitHub)

对于交易机器人来说，最常用的是下面几类：

数据类型	对交易机器人的价值	典型用途
Funding Rate	判断多空成本和拥挤度	过滤追涨杀跌信号
Liquidation History	判断强制平仓方向	识别杠杆清洗
Open Interest	判断资金是否入场	确认趋势强度
Long/Short Ratio	判断账户或大户倾向	情绪辅助指标
Taker Buy/Sell	判断主动买卖力量	短线趋势确认
Order Book	判断流动性和挂单分布	做市、盘口策略
Liquidation Heatmap	判断潜在清算区域	支撑阻力和风控

如果你的交易机器人只需要基础升级，可以先接入：

资金费率
清算历史
持仓量

如果你的机器人面向高频或机构级策略，可以继续接入：

订单簿
主动买卖成交
CVD
清算热力图
WebSocket 实时数据

六、CoinGlass API 接入基础

CoinGlass API V4 的基础路径在公开资料中为：

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

请求时通常需要在 Header 中传入 API Key。公开示例中使用的 Header 名称为：

CG-API-KEY

下面是一个基础 Python 请求结构：

import os
import requests

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

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

def get_data(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()

建议不要把 API Key 直接写在代码里，而是放到环境变量中：

export COINGLASS_API_KEY="your_api_key_here"

七、示例一：获取 BTC 资金费率历史

以下代码演示如何请求 BTC 的资金费率历史数据。具体参数请以 CoinGlass 最新文档为准，因为不同套餐、接口版本和交易所支持范围可能会调整。

import os
import requests
import pandas as pd

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

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

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
    }

    url = f"{BASE_URL}{endpoint}"
    response = requests.get(url, headers=headers, params=params, timeout=10)
    response.raise_for_status()

    return response.json()

data = fetch_funding_rate(symbol="BTC", interval="1h", limit=100)

print(data)

如果接口返回的数据中包含时间、开盘值、最高值、最低值、收盘值等字段，可以进一步整理成 DataFrame：

def normalize_ohlc_response(raw):
    """
    根据实际 API 返回结构调整字段名。
    这里假设 raw["data"] 是列表，每个元素包含 time/open/high/low/close。
    """
    rows = raw.get("data", [])

    df = pd.DataFrame(rows)

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

    return df

funding_raw = fetch_funding_rate("BTC", "1h", 100)
funding_df = normalize_ohlc_response(funding_raw)

print(funding_df.head())

八、示例二：获取 BTC 清算历史

CoinGlass 文档显示，币种聚合清算历史接口路径为：

/api/futures/liquidation/aggregated-history

该接口提供多个交易所中某个币种的多空清算聚合历史数据，支持的参数包括 exchange_list、symbol、interval、limit、start_time、end_time 等。(CoinGlass-API)

Python 示例：

import os
import requests
import pandas as pd

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

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

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
    }

    url = f"{BASE_URL}{endpoint}"
    response = requests.get(url, headers=headers, params=params, timeout=10)
    response.raise_for_status()

    return response.json()

liquidation_raw = fetch_liquidation_history(
    symbol="BTC",
    exchanges="Binance,OKX,Bybit",
    interval="1h",
    limit=100
)

print(liquidation_raw)

整理成 DataFrame：

def normalize_liquidation_response(raw):
    """
    根据实际 API 返回结构调整字段名。
    常见字段可能包括 time、long_liquidation、short_liquidation 等。
    """
    rows = raw.get("data", [])
    df = pd.DataFrame(rows)

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

    return df

liq_df = normalize_liquidation_response(liquidation_raw)

print(liq_df.head())

九、把资金费率和清算数据合并

接下来，我们把资金费率和清算数据合并成一个信号表。

def merge_market_data(funding_df, liq_df):
    """
    将资金费率数据和清算数据按时间合并。
    注意：字段名需要根据实际 API 返回结果调整。
    """

    funding = funding_df.copy()
    liquidation = liq_df.copy()

    funding = funding.sort_values("time")
    liquidation = liquidation.sort_values("time")

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

    return merged

market_df = merge_market_data(funding_df, liq_df)

print(market_df.tail())

假设合并后的数据包含以下字段：

字段	含义
time	时间
funding_close	当前周期资金费率
funding_high	当前周期最高资金费率
funding_low	当前周期最低资金费率
long_liquidation	多头清算金额
short_liquidation	空头清算金额
total_liquidation	总清算金额

如果实际字段不同，可以在数据清洗阶段统一重命名。

market_df = market_df.rename(columns={
    "close": "funding_close",
    "longLiquidation": "long_liquidation",
    "shortLiquidation": "short_liquidation"
})

十、构建一个简单的交易机器人信号模型

下面给出一个示例模型。它不是完整交易策略，而是一个“信号过滤器”。

它的目标是帮助交易机器人判断当前市场环境：

信号	含义
LONG_CROWDED	多头拥挤，谨慎追多
SHORT_CROWDED	空头拥挤，谨慎追空
LONG_LIQUIDATION_SPIKE	多头清算异常
SHORT_LIQUIDATION_SPIKE	空头清算异常
NEUTRAL	市场中性
RISK_OFF	清算异常放大，降低交易频率

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

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

    # 根据实际字段调整
    data["funding_z"] = calculate_zscore(data["funding_close"], window=24)
    data["long_liq_z"] = calculate_zscore(data["long_liquidation"], window=24)
    data["short_liq_z"] = calculate_zscore(data["short_liquidation"], window=24)

    signals = []

    for _, row in data.iterrows():
        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)

        if funding_z > 2 and long_liq_z > 2:
            signal = "LONG_CROWDED_RISK"
        elif funding_z < -2 and short_liq_z > 2:
            signal = "SHORT_SQUEEZE_RISK"
        elif long_liq_z > 3 or short_liq_z > 3:
            signal = "RISK_OFF"
        elif funding_z > 1.5:
            signal = "LONG_CROWDED"
        elif funding_z < -1.5:
            signal = "SHORT_CROWDED"
        else:
            signal = "NEUTRAL"

        signals.append(signal)

    data["derivatives_signal"] = signals

    return data

signal_df = generate_derivatives_signal(market_df)

print(signal_df[["time", "funding_close", "long_liquidation", "short_liquidation", "derivatives_signal"]].tail())

十一、如何接入交易机器人？

假设你的交易机器人原本有一个简单的趋势策略：

def base_trend_signal(price_df):
    """
    简单均线策略示例。
    """
    df = price_df.copy()

    df["ma_fast"] = df["close"].rolling(20).mean()
    df["ma_slow"] = df["close"].rolling(60).mean()

    if df["ma_fast"].iloc[-1] > df["ma_slow"].iloc[-1]:
        return "BUY"
    elif df["ma_fast"].iloc[-1] < df["ma_slow"].iloc[-1]:
        return "SELL"
    else:
        return "HOLD"

现在加入衍生品数据过滤器：

def final_trading_decision(price_signal, derivatives_signal):
    """
    将价格信号和衍生品信号组合。
    """

    # 多头过热时，不追多
    if price_signal == "BUY" and derivatives_signal in [
        "LONG_CROWDED",
        "LONG_CROWDED_RISK",
        "RISK_OFF"
    ]:
        return "HOLD"

    # 空头过热时，不追空
    if price_signal == "SELL" and derivatives_signal in [
        "SHORT_CROWDED",
        "SHORT_SQUEEZE_RISK",
        "RISK_OFF"
    ]:
        return "HOLD"

    # 极端风险环境，降低交易频率
    if derivatives_signal == "RISK_OFF":
        return "HOLD"

    return price_signal

完整调用示例：

price_signal = "BUY"  # 假设来自你的趋势策略
latest_derivatives_signal = signal_df["derivatives_signal"].iloc[-1]

decision = final_trading_decision(
    price_signal=price_signal,
    derivatives_signal=latest_derivatives_signal
)

print("Price Signal:", price_signal)
print("Derivatives Signal:", latest_derivatives_signal)
print("Final Decision:", decision)

这个逻辑的核心不是让机器人“预测未来”，而是让机器人知道什么时候应该少交易。

很多策略亏损并不是因为方向完全错，而是因为在高风险环境下仍然机械执行信号。例如：

原始信号	衍生品环境	优化后动作
BUY	多头资金费率极高	暂停追多
BUY	多头刚发生大量清算	等待确认
SELL	空头资金费率极低	暂停追空
SELL	空头清算异常放大	防止被轧空
BUY/SELL	清算总量异常	降低仓位或暂停交易

十二、加入仓位管理：不只是买卖，还要控制风险

资金费率和清算数据最适合用于 仓位管理。

你可以让交易机器人根据市场风险动态调整仓位：

def calculate_position_size(base_size, derivatives_signal):
    """
    根据衍生品风险信号调整仓位。
    base_size 代表默认仓位，例如 1 表示 100% 标准仓位。
    """

    risk_multiplier = {
        "NEUTRAL": 1.0,
        "LONG_CROWDED": 0.7,
        "SHORT_CROWDED": 0.7,
        "LONG_CROWDED_RISK": 0.4,
        "SHORT_SQUEEZE_RISK": 0.4,
        "RISK_OFF": 0.0
    }

    multiplier = risk_multiplier.get(derivatives_signal, 0.5)

    return base_size * multiplier

base_position_size = 1.0
latest_signal = signal_df["derivatives_signal"].iloc[-1]

position_size = calculate_position_size(base_position_size, latest_signal)

print("Adjusted Position Size:", position_size)

这样一来，交易机器人就不再是简单地“有信号就满仓”，而是会根据衍生品市场风险动态调整。

十三、资金费率 + 清算数据的实战策略框架

下面是一个更完整的策略框架。

模块	输入数据	作用
趋势模块	K 线、均线、价格突破	判断基础方向
情绪模块	资金费率、多空比	判断市场拥挤度
风控模块	清算数据、持仓量变化	判断杠杆风险
执行模块	订单簿、价差、滑点	优化入场和出场
仓位模块	风险信号、波动率	动态控制仓位
复盘模块	历史数据、回测结果	优化参数

一个相对稳健的交易机器人流程可以设计为：

1. 获取价格数据
2. 生成基础趋势信号
3. 获取资金费率数据
4. 获取清算数据
5. 计算市场拥挤度
6. 判断是否存在清算风险
7. 过滤高风险交易信号
8. 动态调整仓位
9. 执行交易
10. 记录结果并回测优化

十四、哪些场景最适合使用 CoinGlass API？

1. 趋势交易机器人

趋势机器人最怕追在顶部或底部。
资金费率可以帮助判断趋势是否过热，清算数据可以帮助判断趋势是否出现杠杆踩踏。

适合用法：

资金费率极高时降低追多权重
空头清算放大时确认上涨动能
多头清算放大时避免盲目抄底
清算后价格收回时寻找反转机会

2. 网格交易机器人

网格机器人最怕单边行情。
清算数据可以帮助识别极端行情，资金费率可以帮助判断市场是否偏向单边。

适合用法：

清算异常放大时暂停新增网格
资金费率持续极端时扩大网格间距
杠杆清洗后恢复正常交易
多空拥挤时降低单侧敞口

3. 套利机器人

资金费率本身就是套利策略的重要变量。
不同交易所之间的资金费率差异，可以形成套利机会。

适合用法：

监控多交易所资金费率
寻找费率异常交易对
结合持仓量判断费率可持续性
结合清算数据判断风险事件

CoinGlass API 文档中资金费率模块包含 Arbitrage 分类，这说明资金费率套利也是其 API 适用场景之一。(GitHub)

4. 风控机器人

风控机器人不一定直接下单，但可以给策略系统提供风险状态。

适合用法：

清算异常提醒
杠杆拥挤提醒
多空挤压提醒
自动降低交易频率
极端行情暂停交易

十五、常见误区

误区一：资金费率高就一定做空

不一定。

强趋势行情中，资金费率可以长时间维持高位。如果只因为资金费率高就做空，很容易逆势亏损。

更合理的做法是：

资金费率高 = 多头拥挤
多头拥挤 ≠ 立刻做空
多头拥挤 + 价格转弱 + 多头清算放大 = 做空条件增强

误区二：出现大量清算就马上反向开仓

也不一定。

大量清算可能是行情结束，也可能只是趋势中继。关键要看清算后价格是否继续破位。

更合理的做法是：

清算后表现	解读
清算后继续破位	趋势可能延续
清算后快速收回	可能是假突破
清算后横盘	等待新方向
清算后资金费率回归中性	杠杆压力下降

误区三：只看单一交易所数据

加密市场是多交易所市场。
如果只看单一交易所，可能会忽略其他主流交易所的杠杆变化。

CoinGlass API 的价值之一在于聚合多个交易所的数据。CoinGlass 官方资料显示，其 API 覆盖 30+ 全球主流加密交易所，并提供统一 API 获取多交易所市场数据。(coinglass)

十六、上线前需要注意什么？

在把资金费率和清算数据接入实盘交易机器人之前，建议完成以下检查：

检查项	说明
数据延迟	确认接口返回是否满足策略频率
字段稳定性	根据文档处理字段变化
错误重试	API 请求失败时自动重试
限速控制	避免请求过于频繁
缓存机制	高频策略应缓存部分数据
回测验证	不要直接实盘上线
风控优先	极端行情下先保护本金
日志记录	记录每次信号和 API 返回

示例：给 API 请求加入重试机制。

import time
import requests

def safe_get(url, headers, params=None, retries=3, sleep_seconds=2):
    last_error = None

    for attempt in range(retries):
        try:
            response = requests.get(
                url,
                headers=headers,
                params=params,
                timeout=10
            )
            response.raise_for_status()
            return response.json()

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

    raise last_error

十七、完整示例：衍生品风险过滤器

下面是一个简化版完整代码，把前面的逻辑组合起来。

import os
import time
import requests
import pandas as pd

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

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

def request_api(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 fetch_funding(symbol="BTC", interval="1h", limit=100):
    endpoint = "/api/futures/fundingRate/oi-weight-ohlc-history"

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

    return request_api(endpoint, params)

def fetch_liquidation(symbol="BTC", interval="1h", limit=100):
    endpoint = "/api/futures/liquidation/aggregated-history"

    params = {
        "exchange_list": "Binance,OKX,Bybit",
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }

    return request_api(endpoint, params)

def to_dataframe(raw):
    rows = raw.get("data", [])
    df = pd.DataFrame(rows)

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

    return df

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

def build_signal_table(funding_df, liq_df):
    funding = funding_df.copy()
    liquidation = liq_df.copy()

    # 以下字段名需要根据真实返回结果调整
    funding = funding.rename(columns={
        "close": "funding_close"
    })

    liquidation = liquidation.rename(columns={
        "longLiquidation": "long_liquidation",
        "shortLiquidation": "short_liquidation"
    })

    funding = funding.sort_values("time")
    liquidation = liquidation.sort_values("time")

    df = pd.merge_asof(
        funding,
        liquidation,
        on="time",
        direction="nearest",
        tolerance=pd.Timedelta("5min")
    )

    df["funding_z"] = zscore(df["funding_close"], 24)
    df["long_liq_z"] = zscore(df["long_liquidation"], 24)
    df["short_liq_z"] = zscore(df["short_liquidation"], 24)

    signals = []

    for _, row in df.iterrows():
        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)

        if funding_z > 2 and long_liq_z > 2:
            signal = "LONG_CROWDED_RISK"
        elif funding_z < -2 and short_liq_z > 2:
            signal = "SHORT_SQUEEZE_RISK"
        elif long_liq_z > 3 or short_liq_z > 3:
            signal = "RISK_OFF"
        elif funding_z > 1.5:
            signal = "LONG_CROWDED"
        elif funding_z < -1.5:
            signal = "SHORT_CROWDED"
        else:
            signal = "NEUTRAL"

        signals.append(signal)

    df["signal"] = signals

    return df

def position_multiplier(signal):
    table = {
        "NEUTRAL": 1.0,
        "LONG_CROWDED": 0.7,
        "SHORT_CROWDED": 0.7,
        "LONG_CROWDED_RISK": 0.4,
        "SHORT_SQUEEZE_RISK": 0.4,
        "RISK_OFF": 0.0
    }

    return table.get(signal, 0.5)

if __name__ == "__main__":
    funding_raw = fetch_funding("BTC", "1h", 100)
    liquidation_raw = fetch_liquidation("BTC", "1h", 100)

    funding_df = to_dataframe(funding_raw)
    liq_df = to_dataframe(liquidation_raw)

    signal_df = build_signal_table(funding_df, liq_df)

    latest = signal_df.iloc[-1]
    latest_signal = latest["signal"]
    multiplier = position_multiplier(latest_signal)

    print("Latest time:", latest["time"])
    print("Derivatives signal:", latest_signal)
    print("Position multiplier:", multiplier)

十八、总结：交易机器人真正需要的是“市场环境感知”

很多交易机器人失败，不是因为代码写得不好，而是因为它们缺少市场环境感知。

价格突破不一定是真突破。
下跌放量不一定是趋势开始。
资金费率高不一定马上下跌。
大量清算也不一定代表行情结束。

但如果你把这些数据组合起来，交易机器人就能更接近真实市场：

数据	解决的问题
价格	现在涨了还是跌了
成交量	交易是否活跃
资金费率	多空哪边更拥挤
清算数据	哪边杠杆正在被清洗
持仓量	是否有新资金进入
多空比	市场账户倾向
订单簿	流动性在哪里
热力图	潜在支撑阻力在哪里