当交易量成为陷阱：衡量“有毒流”与信息流

交易量是最容易展示也最容易误解的指标。

在Polymarket上，高交易量可能意味着：

• 真实信息到来
• 套利和频繁交易
• 机器人试探订单簿
• 交易者追逐噪音而支付价差

如果你的警报仅依赖交易量，你只会变得更嘈杂，而不是更聪明。

解决方法是衡量流质量：

• 信息流：推动价格并维持新价位的交易（新信息）
• 有毒流：看似活跃但导致逆向选择的交易（你刚成交，价格就朝不利方向变动）

内部链接：

• 用回测校准警报：/backtesting
• 跟踪聪明钱包，而非噪音：/smart-money
• 获取高信号警报：/subscribe

---

1) 核心思想：逆向选择

在市场微观结构中，“有毒”流是指让流动性提供者亏损的流：

• 做市商成交
• 中间价随即朝不利方向变动

从跟单者的角度看，有毒流更糟糕：

• 你复制了发起者盈利的交易
• 但你在优势消失后才入场

因此，我们想要衡量：

交易后，中间价在交易方向上移动了多少？

如果价格立即移动并保持，则该流是信息流。 如果价格均值回归或剧烈震荡，则该流主要是噪音。

---

2) 可验证指标：交易后中间价漂移

对于时间 (t) 的每笔交易：

• 定义 (m_t) = 交易时的中间价
• 定义 (m_{t+\Delta}) = (\Delta) 秒后的中间价

对于买入，信息漂移为：

drift = m_{t+\Delta} - m_t

对于卖出，取反：

drift = m_t - m_{t+\Delta}

然后按交易规模归一化，以进行同类比较：

drift_per_$ = drift / trade_notional_usd

如果某个钱包交易的 (drift_per_$) 持续为正，则该钱包的流倾向于信息流。 如果接近零或为负，跟单是危险的。

---

3) 数据获取：交易 + 中间价（公开）

Polymarket 记录了以下公开端点：

• Data API（交易历史）
• CLOB API（中间价和价格历史）
• Gamma API（代币ID和发现） 外部来源：https://docs.polymarket.com/quickstart/reference/endpoints

实用工作流：

1. 选择一个代币ID（YES或NO代币）
2. 获取该代币的近期交易
3. 获取中间价时间序列
4. 在多个时间范围（例如30秒、5分钟、30分钟）计算交易后漂移

---

4) 可复现脚本：代币的有毒流评分

该脚本：

• 获取近期交易（Data API）
• 采样中间价序列（CLOB）
• 在固定时间范围计算交易后漂移

import json
import math
import urllib.request
from statistics import mean

TOKEN_ID = "YOUR_TOKEN_ID"
LIMIT = 200
HORIZONS = [30, 300, 1800]  # 秒

def get_json(url: str):
  with urllib.request.urlopen(url) as r:
    return json.loads(r.read().decode("utf-8"))

trades = get_json(f"https://data-api.polymarket.com/trades?limit={LIMIT}&token_id={TOKEN_ID}")
mid_hist = get_json(f"https://clob.polymarket.com/prices-history?token_id={TOKEN_ID}")

def normalize_points(points):
  out = []
  for row in points:
    ts = int(row.get("t") or row.get("timestamp") or row.get("time"))
    px = float(row.get("p") or row.get("price"))
    out.append((ts, px))
  out.sort(key=lambda x: x[0])
  return out

mid_points = normalize_points(mid_hist)
trade_points = []
for row in trades:
  ts = int(row.get("timestamp") or row.get("time") or row.get("t"))
  price = float(row.get("price"))
  size = float(row.get("size") or row.get("shares") or 0)
  side = (row.get("side") or "").lower()
  trade_usd = float(row.get("trade_usd") or (price * size))
  trade_points.append((ts, side, trade_usd, price))

def mid_at(ts):
  # 向前填充最后一个观测值
  lo, hi = 0, len(mid_points) - 1
  if hi < 0:
    return None
  if ts <= mid_points[0][0]:
    return mid_points[0][1]
  if ts >= mid_points[-1][0]:
    return mid_points[-1][1]
  while lo < hi:
    mid = (lo + hi + 1) // 2
    if mid_points[mid][0] <= ts:
      lo = mid
    else:
      hi = mid - 1
  return mid_points[lo][1]

# 计算每个时间范围的漂移
results = {}
for horizon in HORIZONS:
  drifts = []
  for ts, side, trade_usd, price in trade_points:
    m_t = mid_at(ts)
    m_future = mid_at(ts + horizon)
    if m_t is None or m_future is None:
      continue
    if side == "buy":
      drift = m_future - m_t
    elif side == "sell":
      drift = m_t - m_future
    else:
      continue
    if trade_usd > 0:
      drifts.append(drift / trade_usd)
  if drifts:
    results[horizon] = mean(drifts)
  else:
    results[horizon] = None

print(json.dumps(results, indent=2))

---

5) 解读结果

• 如果所有时间范围的 (drift_per_$) 均为正，则该代币的流主要是信息流。
• 如果短期为正但长期为负，则存在初始推动但随后反转——这是有毒流。
• 如果所有值接近零，则交易是噪音。

实用规则：

• 关注 (drift_per_$) 持续 > 0.0001 的钱包（每1000美元交易推动价格1个基点）。
• 避免 (drift_per_$) 为负的钱包——它们是被逆向选择困住的流动性提供者。
• 使用30秒时间范围进行实时警报；使用30分钟时间范围进行长期信号。

---

下一步

• 在仪表板中集成此指标：/dashboard
• 探索顶级信息交易者：/smart-money
• 设置基于漂移的警报：/alerts