name: 集成 description: “外部数据源、连接器和自定义数据流” emoji: “🔗”

集成 - 完整的API参考

管理外部数据源，添加自定义连接器，并为交易机器人插入新数据流。

聊天命令

列出数据源

/integrations                               列出所有数据源
/integrations status                        显示源健康状态
/integrations sources                       可用源类型

管理源

/integrations enable fedwatch              启用CME FedWatch
/integrations disable 538                   禁用FiveThirtyEight
/integrations add webhook "my-signals"      添加自定义webhook源
/integrations add rest "my-api" <url>       添加REST API源
/integrations remove <source-id>            移除数据源

配置源

/integrations config fedwatch              查看源配置
/integrations set fedwatch interval 60     设置刷新间隔
/integrations set fedwatch key <api-key>   设置API密钥
/integrations test <source-id>             测试源连接

查看数据

/integrations data fedwatch                源的最新数据
/integrations history <source> --hours 24  源的历史数据
/integrations subscribe <source>           实时更新

TypeScript API参考

创建集成管理器

import { createIntegrationsManager } from 'clodds/integrations';

const integrations = createIntegrationsManager({
  // 存储
  storage: 'sqlite',
  dbPath: './integrations.db',

  // 默认刷新间隔
  defaultIntervalMs: 60000,

  // 重试设置
  maxRetries: 3,
  retryDelayMs: 5000,
});

内置数据源

// 启用内置源
await integrations.enable('fedwatch');     // CME FedWatch
await integrations.enable('538');          // FiveThirtyEight
await integrations.enable('silver');       // Silver Bulletin
await integrations.enable('rcp');          // RealClearPolitics
await integrations.enable('odds-api');     // The Odds API
await integrations.enable('polymarket');   // Polymarket价格
await integrations.enable('kalshi');       // Kalshi价格
await integrations.enable('binance');      // Binance现货价格
await integrations.enable('crypto');       // 多交易所加密货币

添加自定义Webhook源

// 添加接收自定义信号的webhook
const source = await integrations.addWebhook({
  name: 'my-signals',
  description: '自定义交易信号',

  // Webhook配置
  path: '/webhooks/my-signals',
  secret: process.env.WEBHOOK_SECRET,

  // 数据模式（可选验证）
  schema: {
    type: 'object',
    properties: {
      signal: { type: 'string', enum: ['BUY', 'SELL', 'HOLD'] },
      symbol: { type: 'string' },
      confidence: { type: 'number', min: 0, max: 1 },
    },
    required: ['signal', 'symbol'],
  },

  // 转换传入数据
  transform: (payload) => ({
    signal: payload.signal,
    symbol: payload.symbol,
    confidence: payload.confidence || 0.5,
    timestamp: Date.now(),
  }),
});

console.log(`Webhook URL: ${source.url}`);
// POST到：https://your-domain.com/webhooks/my-signals

添加自定义REST源

// 添加REST API数据源
const source = await integrations.addRest({
  name: 'my-api',
  description: '自定义价格API',

  // API配置
  url: 'https://api.example.com/prices',
  method: 'GET',
  headers: {
    'Authorization': `Bearer ${process.env.MY_API_KEY}`,
  },

  // 轮询间隔
  intervalMs: 30000,

  // 转换响应
  transform: (response) => ({
    price: response.data.price,
    volume: response.data.volume,
    timestamp: Date.now(),
  }),
});

添加WebSocket源

// 添加WebSocket数据源
const source = await integrations.addWebSocket({
  name: 'live-prices',
  description: '实时价格推送',

  // WebSocket配置
  url: 'wss://stream.example.com/prices',

  // 消息处理器
  onMessage: (data) => ({
    type: 'price',
    symbol: data.s,
    price: parseFloat(data.p),
    timestamp: data.t,
  }),

  // 订阅消息
  subscribe: {
    method: 'SUBSCRIBE',
    params: ['btcusdt@trade'],
  },

  // 重连设置
  reconnect: true,
  reconnectIntervalMs: 5000,
});

订阅数据

// 订阅实时更新
integrations.subscribe('my-signals', (data) => {
  console.log(`信号: ${data.signal} ${data.symbol}`);
  console.log(`置信度: ${data.confidence}`);

  if (data.signal === 'BUY' && data.confidence > 0.8) {
    // 执行交易逻辑
  }
});

// 订阅多个源
integrations.subscribeAll(['fedwatch', 'crypto', 'my-signals'], (source, data) => {
  console.log(`[${source}] ${JSON.stringify(data)}`);
});

获取最新数据

// 从源获取当前数据
const fedData = await integrations.getData('fedwatch');

console.log('联邦利率概率：');
for (const meeting of fedData.meetings) {
  console.log(`${meeting.date}: ${meeting.probabilities}`);
}

// 带新鲜度检查获取
const data = await integrations.getData('crypto', {
  maxAgeMs: 60000,  // 如果超过60秒则重新获取
});

检查状态

// 获取源状态
const status = await integrations.getStatus('my-api');

console.log(`状态: ${status.status}`);  // 'healthy' | 'degraded' | 'error'
console.log(`最后获取: ${status.lastFetch}`);
console.log(`最后错误: ${status.lastError}`);
console.log(`获取次数: ${status.fetchCount}`);
console.log(`错误次数: ${status.errorCount}`);

// 获取所有状态
const all = await integrations.getAllStatuses();

内置数据源

源	类型	数据	刷新
fedwatch	REST	联邦利率概率	5分钟
538	REST	选举预测	1小时
silver	REST	Silver Bulletin预测	1小时
rcp	REST	民调平均值	15分钟
odds-api	REST	体育博彩赔率	1分钟
polymarket	WebSocket	市场价格	实时
kalshi	WebSocket	市场价格	实时
binance	WebSocket	加密货币价格	实时

自定义源类型

类型	最适合	延迟
webhook	外部信号推送给您	即时
rest	您定期轮询的API	秒
websocket	实时流数据	毫秒

在机器人中使用数据

import { createTradingBot } from 'clodds/trading';
import { createIntegrationsManager } from 'clodds/integrations';

const integrations = createIntegrationsManager();
const bot = createTradingBot();

// 在机器人策略中使用自定义信号
integrations.subscribe('my-signals', async (signal) => {
  if (signal.signal === 'BUY' && signal.confidence > 0.9) {
    await bot.execute({
      platform: 'polymarket',
      market: signal.symbol,
      side: 'YES',
      size: 100 * signal.confidence,
    });
  }
});

// 使用联邦数据进行宏观投注
integrations.subscribe('fedwatch', async (data) => {
  const cutProb = data.meetings[0].probabilities['25bp_cut'];
  if (cutProb > 0.8) {
    // 高概率的降息
    await bot.execute({
      platform: 'kalshi',
      market: 'fed-rate-cut',
      side: 'YES',
      size: 500,
    });
  }
});

环境变量

# 内置源
CME_FEDWATCH_API_KEY=your-key
FIVETHIRTYEIGHT_API_KEY=your-key
ODDS_API_KEY=your-key

# 自定义源
MY_SIGNALS_WEBHOOK_SECRET=your-secret
MY_API_KEY=your-key

最佳实践

验证传入数据 — 为webhooks使用模式
设置适当的间隔 — 不要频繁轮询
优雅地处理错误 — 源有时会失败
监控新鲜度 — 对陈旧数据进行警报
一致转换 — 规范化数据格式
使用WebSocket降低延迟 — 当毫秒很重要时