ckvd-usage

用途

CryptoKlineVisionData（简称 CKVD）用于获取加密货币市场行情数据。它通过故障转移控制协议（Failover Control Protocol，FCP）在多个数据源之间自动切换，以提升数据获取的可用性。

主要能力

多数据源自动故障转移，优先级为：缓存（约 1ms）→ Vision API（约 1-5s）→ REST API（约 100-500ms）。
支持的市场类型：SPOT（现货）、FUTURES_USDT（USDT 本位合约）、FUTURES_COIN（币本位合约）。
支持的周期（Interval）包括：MINUTE_1、MINUTE_5、HOUR_1、HOUR_4、DAY_1 等。
符号格式：现货/合约使用 BTCUSDT，币本位合约使用 BTCUSD_PERP。
提供高层 API fetch_market_data，以及管理器对象 CryptoKlineVisionData。
支持实时 K 线流式订阅，包含同步（阻塞迭代）与异步两种方式，返回 KlineUpdate 对象。
附带辅助脚本：符号格式校验、缓存状态检查、FCP 行为诊断。

使用方式

快速开始

from ckvd import CryptoKlineVisionData, DataProvider, MarketType, Interval
from datetime import datetime, timedelta, timezone

# 创建 USDT 本位合约的管理器
manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.FUTURES_USDT)

# 获取数据（自动故障转移：缓存 → Vision → REST）
# 重要：始终使用 UTC 时区感知的 datetime
end_time = datetime.now(timezone.utc)
start_time = end_time - timedelta(days=7)

df = manager.get_data(
    symbol="BTCUSDT",
    start_time=start_time,
    end_time=end_time,
    interval=Interval.HOUR_1
)

print(f"Loaded {len(df)} bars")
manager.close()

高层 API

from ckvd import fetch_market_data, DataProvider, MarketType, Interval, ChartType
from datetime import datetime, timedelta, timezone

df, elapsed_time, records_count = fetch_market_data(
    provider=DataProvider.BINANCE,
    market_type=MarketType.FUTURES_USDT,
    chart_type=ChartType.KLINES,
    symbol="BTCUSDT",
    interval=Interval.HOUR_1,
    start_time=datetime.now(timezone.utc) - timedelta(days=30),
    end_time=datetime.now(timezone.utc)
)
print(f"Loaded {records_count} bars in {elapsed_time:.2f}s")

实时 K 线流式订阅

需先安装流式依赖：pip install crypto-kline-vision-data[streaming]。

同步流式：

from ckvd import CryptoKlineVisionData, DataProvider, MarketType

manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.FUTURES_USDT)

# 阻塞迭代器，接收 KlineUpdate 对象
for update in manager.stream_data_sync("BTCUSDT", "1h"):
    if update.is_closed:
        print(f"Closed candle: open={update.open}, close={update.close}, volume={update.volume}")
    else:
        print(f"Open candle (time: {update.open_time})")

manager.close()

异步流式：

import asyncio
from ckvd import CryptoKlineVisionData, DataProvider, MarketType, StreamConfig

async def main():
    manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.FUTURES_USDT)

    config = StreamConfig(
        market_type=MarketType.FUTURES_USDT,
        confirmed_only=True,  # 仅在 K 线收盘时发出
        queue_maxsize=1000
    )
    stream = manager.create_stream(config)

    async with stream:
        await stream.subscribe("BTCUSDT", "1h")
        async for update in stream:
            print(f"{update.symbol} {update.interval}: close={update.close}")

asyncio.run(main())

KlineUpdate 字段：open_time（UTC datetime）、open、high、low、close、volume（均为 float）、symbol、interval、is_closed（bool）。

辅助脚本

# 校验符号格式
uv run -p 3.13 python docs/skills/ckvd-usage/scripts/validate_symbol.py BTCUSDT FUTURES_COIN

# 检查缓存状态
uv run -p 3.13 python docs/skills/ckvd-usage/scripts/check_cache.py BTCUSDT futures_usdt 1h

# 诊断 FCP 行为（带调试日志）
uv run -p 3.13 python docs/skills/ckvd-usage/scripts/diagnose_fcp.py BTCUSDT futures_usdt 1h --days 3

任务模板

模板 A：为某符号获取数据——根据符号格式确定市场类型，创建管理器，定义 UTC 时间范围，调用 get_data()，校验返回的 DataFrame（非空、索引单调），完成后关闭管理器。
模板 B：调试空结果——校验符号格式，确认时间范围在过去且为 UTC 感知，启用调试日志（CKVD_LOG_LEVEL=DEBUG），用 enforce_source 逐个测试 FCP 数据源，检查缓存目录，记录根因与解决方案。
模板 C：切换市场类型——识别当前市场类型与符号格式，映射到新格式（如 BTCUSDT → BTCUSD_PERP），创建新管理器并验证数据获取，更新依赖代码。
模板 D：流式获取实时数据——判断同步或异步需求，创建管理器，同步用 stream_data_sync，异步用 StreamConfig + create_stream + subscribe，处理 KlineUpdate 并检查 is_closed，完成后关闭管理器。

适用场景

量化研究与行情回测：批量拉取历史 K 线数据。
实时行情监控：通过流式 API 订阅实时 K 线更新。
多市场数据获取：在现货、USDT 本位合约、币本位合约之间切换。

限制

时间参数必须使用 UTC 时区感知的 datetime。
流式功能需额外安装 streaming 依赖。
关于收费、价格、平台支持范围等具体信息，资料中暂未提供。

详细参考

@references/market-types.md - 市场类型详细文档
@references/intervals.md - 完整周期参考
@references/fcp-protocol.md - FCP 架构与调试
@references/debugging.md - 调试技巧与故障排查
@references/evolution-log.md - 技能变更历史