メニュー
Fetch market data using CryptoKlineVisionData with Failover Control Protocol (cache → Vision API → REST API). TRIGGERS - fetch market data, use CKVD, access Binance, get klines, OHLCV data, CryptoKlineVisionData API.
Codex または Claude でインストール この Prompt をコピーして Codex、Claude、または他のアシスタントに貼り付けると、Skill ページを確認してインストールできます。
SOC 職業分類に基づく
| name | ckvd-usage |
| description | Fetch market data using CryptoKlineVisionData with Failover Control Protocol (cache → Vision API → REST API). TRIGGERS - fetch market data, use CKVD, access Binance, get klines, OHLCV data, CryptoKlineVisionData API. |
| argument-hint | [symbol] [market-type] |
| user-invocable | true |
| allowed-tools | Read, Bash |
Fetch market data for: $ARGUMENTS
Use automatic failover between data sources.
from ckvd import CryptoKlineVisionData, DataProvider, MarketType, Interval
from datetime import datetime, timedelta, timezone
# Create manager for USDT-margined futures
manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.FUTURES_USDT)
# Fetch data with automatic failover (cache → Vision → REST)
# IMPORTANT: Always use UTC timezone-aware datetimes
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()
| Concept | Quick Reference |
|---|---|
| Market Types | SPOT, FUTURES_USDT, FUTURES_COIN |
| Intervals | MINUTE_1, MINUTE_5, HOUR_1, HOUR_4, DAY_1 |
| FCP Priority | Cache (~1ms) → Vision (~1-5s) → REST (~100-500ms) |
| Symbol Format | BTCUSDT (spot/futures), BTCUSD_PERP (coin-margined) |
| Streaming | Real-time klines (sync/async), KlineUpdate objects |
For simpler use cases, use fetch_market_data:
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")
For real-time market data, use the streaming API (requires pip install crypto-kline-vision-data[streaming]):
from ckvd import CryptoKlineVisionData, DataProvider, MarketType
manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.FUTURES_USDT)
# Blocking iterator - receives KlineUpdate objects
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)
# Create stream with optional config
config = StreamConfig(
market_type=MarketType.FUTURES_USDT,
confirmed_only=True, # Only emit when candle closes
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 fields: open_time (UTC datetime), open, high, low, close, volume (all float), symbol, interval, is_closed (bool)
Practical code examples:
Utility scripts for common operations:
# Validate symbol format
uv run -p 3.13 python docs/skills/ckvd-usage/scripts/validate_symbol.py BTCUSDT FUTURES_COIN
# Check cache status
uv run -p 3.13 python docs/skills/ckvd-usage/scripts/check_cache.py BTCUSDT futures_usdt 1h
# Diagnose FCP behavior (with debug logging)
uv run -p 3.13 python docs/skills/ckvd-usage/scripts/diagnose_fcp.py BTCUSDT futures_usdt 1h --days 3
1. Determine market type from symbol format (BTCUSDT vs BTCUSD_PERP)
2. Create CryptoKlineVisionData manager with correct MarketType
3. Define UTC time range with datetime.now(timezone.utc)
4. Call get_data() with symbol, interval, and time range
5. Validate returned DataFrame (not empty, monotonic index)
6. Close manager when done
1. Validate symbol format for market type (validate_symbol_for_market_type)
2. Verify time range is in the past and UTC-aware
3. Enable debug logging (CKVD_LOG_LEVEL=DEBUG)
4. Test each FCP source individually with enforce_source
5. Check cache directory exists and has data
6. Document root cause and resolution
1. Identify current market type and symbol format
2. Map symbol to new market type format (e.g., BTCUSDT -> BTCUSD_PERP)
3. Create new manager with target MarketType
4. Verify data fetches correctly for new market type
5. Update any dependent code with new symbol format
1. Determine if sync (blocking) or async streaming is needed
- Sync: Simple use case, single symbol, main thread OK
- Async: Multiple symbols, integration with other async code
2. Create CryptoKlineVisionData manager
3. For sync: Use manager.stream_data_sync("SYMBOL", "INTERVAL")
4. For async: Create StreamConfig, create_stream(), subscribe to symbols
5. Process KlineUpdate objects in loop
6. Check is_closed flag to detect candle completion
7. Close manager when done (use context manager for async)
After modifying this skill:
For deeper information, see:
Run tests for crypto-kline-vision-data with proper markers and coverage. TRIGGERS - write tests, run tests, pytest, test coverage, unit tests, integration tests, mocking patterns.
Research CKVD codebase for implementation details, patterns, or architecture questions. TRIGGERS - how does FCP work, understand data flow, find code, explore codebase, architecture questions.
Monitor and diagnose FCP (Failover Control Protocol) behavior, cache health, and data source performance. TRIGGERS - FCP issues, cache problems, slow data, Vision API errors, REST fallback, data source debugging.