3. 行情数据源解析：常见行情数据源及其特点

在构建行情数据坏账率哨兵系统时，数据源的选择直接决定了模型的覆盖范围、时效性与准确性。不同数据源在数据格式、更新频率、可信度与获取成本上存在显著差异。本章将系统解析三类核心行情数据源：交易所数据、OTC（场外交易）数据与链上数据，并对比其适用场景与风险特征。

3.1 交易所数据：高频、结构化、中心化

交易所数据是哨兵系统最基础的数据源，通常通过 REST API 或 WebSocket 获取。其特点是数据标准化程度高、延迟低、历史数据完整，适合用于实时监控与回测分析。

3.2 OTC 数据：低频、大额、非标准化

OTC 数据主要来自场外交易平台（如 Cumberland、Genesis）或机构询价系统。其特点是单笔金额大、流动性分散、报价存在价差，且数据通常以 CSV 或邮件形式提供，缺乏统一的 API 接口。

• 数据特征： 报价频率低（分钟级或小时级），买卖价差可达交易所的 2-5 倍。
• 适用场景： 大额交易对市场冲击的评估、机构级坏账率模型中的流动性折价因子。
• 数据清洗难点： 需要手动处理缺失值、剔除重复报价、对齐时间戳。
• 坏账关联： OTC 市场的信息不对称性更高，容易隐藏“假成交”或“对敲”行为，是坏账率哨兵重点监控的盲区。

3.3 链上数据：透明、不可篡改、高延迟

链上数据直接来源于区块链节点（如 Ethereum、Solana 的全节点），包含交易哈希、转账金额、合约调用、Gas 费用等。其最大优势是数据不可篡改，但存在区块确认延迟（通常 12-60 秒）与存储冗余问题。

3.4 三类数据源对比总览

以下表格从哨兵系统最关心的四个维度进行横向对比：

3.5 数据源架构与哨兵系统集成

下图展示了一个典型的哨兵系统数据接入层架构，三类数据源经过适配器（Adapter）统一转换为内部标准格式，再进入坏账率计算引擎。

3.6 数据源选择策略与代码示例

在实际项目中，建议按照以下优先级选择数据源：

1. 实时监控场景： 交易所 WebSocket 数据（首选） + 链上事件订阅（辅助）。
2. 回测与模型训练： 交易所历史 K 线数据（CSV 批量下载） + 链上历史交易记录。
3. 大额交易风控： OTC 报价数据（单独通道） + 链上大额转账监控。

以下是一个使用 Python 同时接入交易所 REST API 与链上 WebSocket 的示例框架：

import asyncio
import aiohttp
from web3 import Web3

# 交易所数据接入（Binance 示例）
async def fetch_exchange_kline(symbol="BTCUSDT", interval="1m"):
    url = f"https://api.binance.com/api/v3/klines?symbol={symbol}&interval={interval}&limit=10"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            data = await resp.json()
            # 返回 [timestamp, open, high, low, close, volume]
            return [[float(x[0]), float(x[1]), float(x[2]), float(x[3]), float(x[4]), float(x[5])] for x in data]

# 链上数据接入（Ethereum 示例）
def fetch_onchain_transactions(w3, block_number):
    block = w3.eth.get_block(block_number, full_transactions=True)
    txs = []
    for tx in block.transactions:
        txs.append({
            "hash": tx.hash.hex(),
            "from": tx["from"],
            "to": tx.to,
            "value": tx.value,
            "gas_price": tx.gasPrice
        })
    return txs

# 哨兵数据统一入口
async def sentinel_data_pipeline():
    # 并行获取交易所与链上数据
    kline_task = fetch_exchange_kline()
    w3 = Web3(Web3.HTTPProvider("https://mainnet.infura.io/v3/YOUR_PROJECT_ID"))
    onchain_task = asyncio.to_thread(fetch_onchain_transactions, w3, w3.eth.block_number)
    
    kline_data, onchain_data = await asyncio.gather(kline_task, onchain_task)
    return {"exchange": kline_data, "onchain": onchain_data}