# Ckvd Usage

> 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. Use when this capability is needed.

- Skill: `tomevault-io/ckvd-usage` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add tomevault-io/ckvd-usage`
- Raw SKILL.md: https://api.skillmd.com/api/skills/tomevault-io/ckvd-usage/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Integrations & APIs
- Author: tomevault-io (https://skillmd.com/u/tomevault-io)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/tomevault-io/ckvd-usage

---


# CryptoKlineVisionData Usage

Fetch market data for: $ARGUMENTS

Use automatic failover between data sources.

## Quick Start

```python
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

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

## High-Level API

For simpler use cases, use `fetch_market_data`:

```python
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

```python
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

```python
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:

```bash
# 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](./references/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](https://tomevault.io/claim/terrylica) — claim your Tome and manage your conversions.
<!-- tomevault:4.0:skill_md:2026-04-12 -->