Ckvd Usage

CryptoKlineVisionData Usage

Fetch market data for: $ARGUMENTS

Use automatic failover between data sources.

Quick Start

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()

Key Concepts

High-Level API

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")

Streaming Real-Time Klines

For real-time market data, use the streaming API (requires pip install crypto-kline-vision-data[streaming]):

Synchronous 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()

Asynchronous Streaming

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)

Examples

Practical code examples:

• @examples/basic-fetch.md - Single/multiple symbols, market types
• @examples/error-handling.md - Exception patterns, validation, retries

Helper Scripts

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

TodoWrite Task Templates

Template A: Fetch Data for Symbol

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

Template B: Debug Empty Results

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

Template C: Switch Market Type

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

Template D: Stream Real-Time Data

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)

Post-Change Checklist

After modifying this skill:

• Quick Start code example still matches actual API
• Helper script commands are correct and scripts exist
• @-links to references/ and examples/ resolve
• Append changes to evolution-log.md

Detailed References

For deeper information, see:

• @references/market-types.md - Detailed market type documentation
• @references/intervals.md - Complete interval reference
• @references/fcp-protocol.md - FCP architecture and debugging
• @references/debugging.md - Debugging techniques and troubleshooting
• @references/evolution-log.md - Skill change history

Converted and distributed by TomeVault — claim your Tome and manage your conversions.