Crypto Kline Vision Data

High-performance market data integration with Failover Control Protocol (FCP).

Package: crypto-kline-vision-data | Import: ckvd | Python: 3.13

Features

• Failover Control Protocol (FCP): Cache (~1ms) → Vision API (~1-5s) → REST API (~100-500ms) — automatic failover, retry, and gap detection
• Apache Arrow Cache: Memory-mapped local cache for instant repeated access
• Binance Vision API: Bulk historical data from AWS S3 (no rate limits, ~48h delay)
• Binance REST API: Real-time data with built-in rate limit handling (Spot, USDS-M Futures, Coin-M Futures)
• WebSocket Streaming (optional): Real-time kline updates via KlineStream with automatic reconnect and state machine (install with [streaming] extra)
• Polars Engine: Internal Polars LazyFrames + streaming; pandas or Polars output at API boundary
• AI Agent Introspection: __probe__.py module for stateless API discovery
• Security: Symbol validation (CWE-22 path traversal prevention)
• Machine-Parseable Errors: All exceptions carry .details dict

Quick Start

git clone https://github.com/terrylica/crypto-kline-vision-data.git
cd crypto-kline-vision-data
uv sync --dev

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

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

end = datetime.now(timezone.utc)
start = end - timedelta(days=7)

df = manager.get_data("BTCUSDT", start, end, Interval.HOUR_1)
print(f"Loaded {len(df)} bars")
manager.close()

Or as a Git dependency in your pyproject.toml:

dependencies = [
    "crypto-kline-vision-data @ git+https://github.com/terrylica/crypto-kline-vision-data.git"
]

Installation

Core Package (Historical & REST Data)

pip install crypto-kline-vision-data
# or
uv add crypto-kline-vision-data

With Streaming (Optional WebSocket Real-Time)

pip install crypto-kline-vision-data[streaming]
# or
uv add crypto-kline-vision-data[streaming]

Streaming requires: websockets>=16.0, orjson>=3.10, python-statemachine>=2.4, stamina>=24.3

For Claude Code Users

This repository is optimized for Claude Code with a hub-and-spoke CLAUDE.md architecture. When Claude Code opens any directory, it automatically loads that directory's CLAUDE.md — giving it domain-specific context, conventions, and deep links to related documentation.

Start here: CLAUDE.md (root hub) — then Claude Code discovers everything else autonomously.

Hub-and-Spoke Architecture

CLAUDE.md (root hub)
├── src/CLAUDE.md         → Package structure, FCP, exceptions, __probe__, security
├── tests/CLAUDE.md       → Test commands, markers, fixtures, mocking patterns
├── docs/CLAUDE.md        → ADRs, skills, benchmarks, troubleshooting
├── examples/CLAUDE.md    → Example conventions, NDJSON telemetry schema
├── scripts/CLAUDE.md     → Dev scripts, mise tasks, cache tools
└── playground/CLAUDE.md  → Experimental prototypes

Each spoke contains only the context relevant to that directory. Claude Code loads them on demand — no context window waste.

What Claude Code Gets

• Skills in docs/skills/ — progressive disclosure guides for usage, testing, research, FCP monitoring
• Agents in .claude/agents/ — specialized subagents (API reviewer, test writer, FCP debugger, silent failure hunter)
• Commands in .claude/commands/ — /review-ckvd, /feature-dev
• Full API surface via from ckvd.__probe__ import discover_api — JSON-serializable metadata for agent introspection

Tips for Working with Claude Code

1. Just ask — Claude Code reads the relevant CLAUDE.md files automatically when you work in a directory
2. Use skills — ask Claude to "fetch BTCUSDT data" or "run tests" and it discovers the right patterns
3. Use agents — @silent-failure-hunter to audit code, @test-writer to generate tests
4. Use probe — from ckvd.__probe__ import discover_api for programmatic API discovery

Examples

# Via mise tasks
mise run demo:quickstart          # Minimal FCP usage
mise run demo:features            # Feature engineering pipeline
mise run demo:cache               # Cache toggle mechanisms
mise run demo:logging             # Logging configuration
mise run demo:datetime            # Timezone handling
mise run demo:one-second          # 1s interval (SPOT only)
mise run demo:lazy                # Lazy initialization

# Or directly
uv run -p 3.13 python examples/quick_start.py

All examples emit NDJSON telemetry to examples/logs/events.jsonl. See examples/CLAUDE.md for schema and parsing.

API Reference

Core API

from ckvd import CryptoKlineVisionData, DataProvider, MarketType, Interval

# Manager-based (recommended)
manager = CryptoKlineVisionData.create(DataProvider.BINANCE, MarketType.FUTURES_USDT)
df = manager.get_data("BTCUSDT", start, end, Interval.HOUR_1)
manager.close()

# High-level function
from ckvd import fetch_market_data, ChartType
df, elapsed, count = fetch_market_data(
    provider=DataProvider.BINANCE, market_type=MarketType.SPOT,
    chart_type=ChartType.KLINES, symbol="BTCUSDT",
    interval=Interval.HOUR_1, start_time=start, end_time=end
)

Market Types and Symbols

Output Formats

# Default: pandas DataFrame (backward compatible)
df = manager.get_data("BTCUSDT", start, end, Interval.HOUR_1)

# Opt-in: Polars DataFrame (zero-copy, faster)
df = manager.get_data("BTCUSDT", start, end, Interval.HOUR_1, return_polars=True)

WebSocket Streaming (Optional)

Real-time kline updates via WebSocket. Requires pip install crypto-kline-vision-data[streaming].

Sync iteration:

from ckvd import CryptoKlineVisionData, DataProvider, MarketType

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

# Synchronous iterator — internally starts a WebSocket in a background thread
for update in manager.stream_data_sync("BTCUSDT", "1h"):
    print(f"{update.symbol}: {update.close}")
    if update.is_closed:
        print("  → Candle closed")

manager.close()

Async iteration:

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

    async with manager.create_stream(confirmed_only=True) as stream:
        await stream.subscribe("BTCUSDT", "1h")
        async for update in stream:
            print(f"{update.symbol}: {update.close} ({update.volume} vol)")

    manager.close()

Streaming features:

• Automatic reconnect: State machine handles connection drops
• KlineUpdate: Frozen dataclass with symbol, open/high/low/close/volume, open_time, is_closed
• Rate-limited: Respects Binance connection limits (10 concurrent streams)

Error Handling

from ckvd.utils.for_core.rest_exceptions import RateLimitError, RestAPIError
from ckvd.utils.for_core.vision_exceptions import VisionAPIError

try:
    df = manager.get_data(...)
except RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s. Details: {e.details}")
except (RestAPIError, VisionAPIError) as e:
    print(f"Error: {e}. Details: {e.details}")

Streaming API Types

Available when installing with [streaming] extra:

from ckvd import KlineUpdate, KlineStream

# KlineUpdate: Frozen dataclass (real-time candle update)
# - symbol: str
# - interval: str (e.g. "1h")
# - open_time: datetime (UTC, start of candle)
# - open, high, low, close: float
# - volume: float
# - is_closed: bool (True when k.x=True, candle is finalized)
# - data_source: str (always "STREAMING")

# Async streaming (KlineStream):
#   async with manager.create_stream(confirmed_only=True, queue_maxsize=1000) as stream:
#       await stream.subscribe("BTCUSDT", "1h")
#       async for update in stream:
#           print(update.close)
#
# Sync streaming (iterator):
#   for update in manager.stream_data_sync("BTCUSDT", "1h"):
#       print(update.close)

# create_stream() kwargs: confirmed_only (default True), queue_maxsize (default 1000),
#                         max_reconnect_attempts (default 5)

Environment Variables

Development

uv sync --dev                    # Install dependencies
mise trust                       # Load environment
uv run -p 3.13 pytest tests/unit/ -v   # Run tests (900 passing)
uv run -p 3.13 ruff check --fix .      # Lint

See CLAUDE.md for full development conventions, commit trailers, and release workflow.

Documentation

License

MIT License — See LICENSE file for details.