You have built a dashboard monitoring 100 US stocks with minute-level OHLCV data. It refreshes every 60 seconds. The engineering works perfectly. Then your finance team asks the question that stops every quant startup: what does this actually cost?
Most developers discover the answer by receiving a billing alert. By then, the damage is done. The goal of this article is to prevent that conversation entirely. We will walk through a rigorous cost estimation framework for TickDB usage, demonstrate the exact API call patterns that drive monthly bills, and provide production-grade code that cuts consumption by 60–80% without sacrificing data fidelity.
This is not theoretical. We will use real formulas, run real calculations, and ship code you can deploy today.
Understanding the TickDB Pricing Model
Before optimizing anything, you need to understand what you are paying for. TickDB pricing operates on a call-based model where each API request incurs a cost unit. The specifics determine your optimization strategy.
The Three Pillars of TickDB Billing
| Cost Component | Description | Impact on Your Dashboard |
|---|---|---|
| Request count | Each HTTP request or WebSocket message counts as one or more units | Every poll cycle = N requests |
| Data volume | Some endpoints charge based on returned data size | Depth channel, multi-symbol queries |
| Plan tier | Free, Professional, Enterprise with different rate limits and per-unit pricing | Determines your baseline cost floor |
The free tier provides approximately 10,000 call units per month — enough for light experimentation but insufficient for production monitoring. Professional plans scale linearly with usage, making per-call optimization directly translate to dollar savings.
Rate Limit Implications
TickDB enforces rate limits through error code 3001. When you exceed your plan's rate limit, the API returns a Retry-After header instructing you to wait before retrying. Ignoring this header does not save you money; it generates failed requests that still count toward your quota in some configurations.
Understanding rate limits is essential because aggressive polling does not deliver more data. It delivers the same data slower, plus error handling overhead.
The Cost Estimation Framework
With the pricing model mapped, we can now build a rigorous estimation formula. This is where most articles fail — they give you a rough guess. We will give you a formula you can plug into a spreadsheet and verify against your actual usage.
Baseline Scenario: Naive Polling
Consider your scenario: 100 stocks, minute-level data, refreshed every 60 seconds.
The naive implementation makes one API call per stock per refresh cycle:
Calls per minute = 100 stocks × 1 call/stock = 100 calls/minute
Calls per hour = 100 × 60 = 6,000 calls/hour
Calls per day = 100 × 60 × 24 = 144,000 calls/day
Calls per month = 144,000 × 30 = 4,320,000 calls/month
At $0.0001 per call unit (illustrative Professional tier rate), this scenario costs:
Monthly cost = 4,320,000 × $0.0001 = $432/month
That number should stop you in your tracks. Polling 100 stocks naively is not a dashboard. It is a budget crisis.
Refined Scenario: Batch Requests
TickDB supports batch endpoints that return data for multiple symbols in a single request. The practical limit is typically 50 symbols per request for most endpoints.
Requests per minute = ceil(100 stocks / 50 per request) = 2 requests/minute
Requests per hour = 2 × 60 = 120 requests/hour
Requests per day = 2 × 60 × 24 = 2,880 requests/day
Requests per month = 2,880 × 30 = 86,400 requests/month
Monthly cost with batching:
Monthly cost = 86,400 × $0.0001 = $8.64/month
Batching alone delivers a 98% reduction in API calls. The remaining 1.6% of the original cost is the baseline for further optimization.
The Cost Estimation Formula
Generalize this into a reusable formula:
Monthly Calls = (S / B) × (60 / I) × 60 × 24 × 30
Where:
S = Number of symbols being monitored
B = Batch size per request (max recommended: 50)
I = Polling interval in minutes
Your monthly cost:
Monthly Cost = Monthly Calls × Cost Per Call
Comparison Table: Naive vs. Optimized
| Approach | Calls/Month | Relative Cost | Latency Risk |
|---|---|---|---|
| Naive polling (1 symbol/call) | 4,320,000 | 100% | None |
| Batch requests (50/call) | 86,400 | 2% | Low |
| Cache-first with 60s TTL | ~86,400 + cache misses | 2–8% | Minimal |
| WebSocket real-time push | ~43,200 (heartbeat included) | 1% | None |
| Hybrid (cache + WebSocket) | ~43,200 + sparse REST | 1–3% | None |
Production-Grade Cost Optimization Code
Theory is insufficient without implementation. We will walk through three optimization layers, each building on the last.
Layer 1: Batch Request Client with Intelligent Caching
The foundation of any cost-optimized TickDB integration is a client that batches requests and caches responses intelligently.
import os
import time
import requests
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import hashlib
import threading
@dataclass
class CacheEntry:
"""Single cache entry with TTL support."""
data: Dict
timestamp: datetime
ttl_seconds: int
def is_expired(self) -> bool:
return datetime.now() - self.timestamp > timedelta(seconds=self.ttl_seconds)
class TickDBOptimizedClient:
"""
TickDB client with batch request support and TTL-based caching.
Engineering notes:
- Thread-safe cache with lock contention acceptable for dashboard use cases.
- TTL of 60 seconds aligns with minute-level data refresh requirements.
- Batch size of 50 maximizes API efficiency per call.
"""
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = "https://api.tickdb.ai/v1",
batch_size: int = 50,
cache_ttl: int = 60,
):
self.api_key = api_key or os.environ.get("TICKDB_API_KEY")
if not self.api_key:
raise ValueError(
"TickDB API key required. Set TICKDB_API_KEY environment variable "
"or pass api_key parameter."
)
self.base_url = base_url.rstrip("/")
self.batch_size = batch_size
self.cache_ttl = cache_ttl
self._cache: Dict[str, CacheEntry] = {}
self._cache_lock = threading.Lock()
self._request_count = 0 # For usage tracking
def _generate_cache_key(self, endpoint: str, params: Dict) -> str:
"""Generate deterministic cache key from endpoint and parameters."""
param_str = "|".join(f"{k}={sorted(v) if isinstance(v, list) else v}"
for k, v in sorted(params.items()))
raw = f"{endpoint}:{param_str}"
return hashlib.md5(raw.encode()).hexdigest()
def _get_cached(self, cache_key: str) -> Optional[Dict]:
"""Retrieve cached response if valid. Thread-safe."""
with self._cache_lock:
entry = self._cache.get(cache_key)
if entry and not entry.is_expired():
return entry.data
elif entry:
del self._cache[cache_key]
return None
def _set_cached(self, cache_key: str, data: Dict) -> None:
"""Store response in cache. Thread-safe."""
with self._cache_lock:
self._cache[cache_key] = CacheEntry(
data=data,
timestamp=datetime.now(),
ttl_seconds=self.cache_ttl
)
def _batch_symbols(self, symbols: List[str]) -> List[List[str]]:
"""Split symbol list into batches of max batch_size."""
return [symbols[i:i + self.batch_size]
for i in range(0, len(symbols), self.batch_size)]
def _make_request(
self,
endpoint: str,
params: Dict,
cache_key: Optional[str] = None,
) -> Dict:
"""
Execute single API request with timeout and error handling.
⚠️ Engineering warning: This method does not implement retry logic.
For production dashboards, wrap this in exponential backoff (see Layer 3).
"""
# Check cache first
if cache_key:
cached = self._get_cached(cache_key)
if cached is not None:
return {"cached": True, "data": cached}
headers = {"X-API-Key": self.api_key}
url = f"{self.base_url}/{endpoint}"
# ⚠️ CRITICAL: Always set timeouts. Never leave requests hanging.
response = requests.get(
url,
headers=headers,
params=params,
timeout=(3.05, 10) # (connect_timeout, read_timeout)
)
self._request_count += 1
if response.status_code == 200:
data = response.json()
if cache_key:
self._set_cached(cache_key, data)
return {"cached": False, "data": data}
# Handle rate limiting
if response.status_code == 429 or (response.is_json and
response.json().get("code") == 3001):
retry_after = int(response.headers.get("Retry-After", 5))
raise RateLimitError(
f"Rate limit exceeded. Retry after {retry_after} seconds.",
retry_after=retry_after
)
response.raise_for_status()
raise ValueError(f"Unexpected response: {response.status_code}")
def get_klines_batch(
self,
symbols: List[str],
interval: str = "1m",
limit: int = 1,
) -> Dict[str, Dict]:
"""
Fetch latest kline data for multiple symbols using batch requests.
Returns:
Dict mapping symbol -> kline data
"""
results = {}
batches = self._batch_symbols(symbols)
for batch in batches:
params = {
"symbol": ",".join(batch),
"interval": interval,
"limit": limit,
}
cache_key = self._generate_cache_key("market/kline/latest", params)
result = self._make_request("market/kline/latest", params, cache_key)
# Parse response and map to symbols
data = result["data"]
if "data" in data:
for item in data["data"]:
symbol = item.get("symbol", item.get("s"))
results[symbol] = item
return results
def get_usage_stats(self) -> Dict:
"""Return current request count for monitoring."""
return {
"total_requests": self._request_count,
"cache_entries": len(self._cache),
}
class RateLimitError(Exception):
"""Raised when API rate limit is exceeded."""
def __init__(self, message: str, retry_after: int):
super().__init__(message)
self.retry_after = retry_after
Layer 2: Cache-First Polling with TTL Enforcement
The batch client above reduces calls dramatically, but we can do better. The principle: never call the API if you have fresh enough data.
import schedule
import time
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CostAwareScheduler:
"""
Scheduler that respects cache TTL and only polls when necessary.
Key insight: If cache TTL is 60 seconds, polling every 30 seconds
is wasteful. Poll every 55 seconds (10% buffer for network variance).
"""
def __init__(
self,
tickdb_client: TickDBOptimizedClient,
symbols: List[str],
poll_interval: int = 55, # Slightly less than cache TTL
):
self.client = tickdb_client
self.symbols = symbols
self.poll_interval = poll_interval
def refresh_data(self) -> Dict[str, Dict]:
"""
Fetch latest data, respecting cache TTL.
Returns data only if cache is stale. This is the core cost-saving mechanism.
"""
start_time = time.time()
try:
data = self.client.get_klines_batch(self.symbols)
elapsed = time.time() - start_time
stats = self.client.get_usage_stats()
logger.info(
f"Refresh complete: {len(self.symbols)} symbols, "
f"{elapsed:.2f}s, {stats['total_requests']} total requests"
)
return data
except RateLimitError as e:
logger.warning(f"Rate limit hit, waiting {e.retry_after}s: {e}")
time.sleep(e.retry_after)
return self.refresh_data()
def run(self, duration_seconds: Optional[int] = None):
"""
Run the polling loop.
Args:
duration_seconds: None for infinite loop, or seconds to run
"""
end_time = time.time() + duration_seconds if duration_seconds else None
logger.info(
f"Starting cost-aware scheduler: {len(self.symbols)} symbols, "
f"{self.poll_interval}s interval"
)
while True:
if end_time and time.time() >= end_time:
break
self.refresh_data()
time.sleep(self.poll_interval)
# Usage example for 100-stock dashboard
if __name__ == "__main__":
# Sample list of 100 US stocks (abbreviated for example)
US_STOCKS = [
"AAPL.US", "MSFT.US", "GOOGL.US", "AMZN.US", "NVDA.US",
"META.US", "TSLA.US", "BRK.B.US", "JPM.US", "V.US",
# ... (90 more symbols)
] * 10 # Simulate 100 unique symbols
US_STOCKS = list(set(US_STOCKS))[:100] # Ensure 100 unique
client = TickDBOptimizedClient(batch_size=50, cache_ttl=60)
scheduler = CostAwareScheduler(
tickdb_client=client,
symbols=US_STOCKS,
poll_interval=55, # Polls once per minute, cache TTL aligned
)
# Run for 1 hour to estimate daily usage
scheduler.run(duration_seconds=3600)
Layer 3: WebSocket Real-Time Alternative
For applications requiring sub-second updates, WebSocket connections eliminate polling overhead entirely. A single persistent connection delivers all symbol updates.
import json
import time
import threading
import websocket
from typing import Callable, Optional, List
class TickDBWebSocketClient:
"""
WebSocket client for real-time TickDB data.
Engineering notes:
- Single connection handles unlimited symbols.
- No polling required — server pushes updates.
- Heartbeat (ping/pong) maintains connection health.
- Automatic reconnection with exponential backoff.
⚠️ For production HFT workloads, migrate to aiohttp/asyncio
for non-blocking I/O. This synchronous implementation suits
dashboard and research use cases.
"""
def __init__(
self,
api_key: str,
on_message: Callable[[dict], None],
on_error: Optional[Callable[[Exception], None]] = None,
):
self.api_key = api_key
self.on_message = on_message
self.on_error = on_error or (lambda e: print(f"WebSocket error: {e}"))
self.ws: Optional[websocket.WebSocketApp] = None
self._running = False
self._reconnect_delay = 1
self._max_reconnect_delay = 60
def connect(self, symbols: List[str], channels: List[str] = None):
"""
Establish WebSocket connection with symbol subscription.
⚠️ Authentication via URL parameter, not header.
"""
if channels is None:
channels = ["kline"] # Real-time candle updates
symbols_param = ",".join(symbols)
# WebSocket auth: API key in URL parameter
url = (
f"wss://api.tickdb.ai/ws"
f"?api_key={self.api_key}"
f"&symbol={symbols_param}"
f"&channel={','.join(channels)}"
)
self.ws = websocket.WebSocketApp(
url,
on_message=self._handle_message,
on_error=self._handle_error,
on_close=self._handle_close,
on_open=self._handle_open,
)
self._running = True
self._thread = threading.Thread(target=self.ws.run_forever)
self._thread.daemon = True
self._thread.start()
print(f"WebSocket connected: {len(symbols)} symbols on channels {channels}")
def _handle_open(self, ws):
"""Send ping immediately after connection established."""
ws.send(json.dumps({"cmd": "ping"}))
print("WebSocket connection opened, heartbeat sent")
def _handle_message(self, ws, message):
"""Process incoming data and invoke callback."""
try:
data = json.loads(message)
# Handle pong response
if data.get("cmd") == "pong":
return
self.on_message(data)
except json.JSONDecodeError as e:
print(f"Failed to parse message: {e}")
def _handle_error(self, ws, error):
"""Log error and prepare for reconnection."""
self.on_error(error)
self._running = False
def _handle_close(self, ws, close_status_code, close_msg):
"""Attempt reconnection with exponential backoff + jitter."""
print(f"WebSocket closed: {close_status_code} - {close_msg}")
self._running = False
# Exponential backoff with jitter
delay = self._reconnect_delay
jitter = delay * 0.1 * (time.time() % 1) # Pseudo-random jitter
actual_delay = min(delay + jitter, self._max_reconnect_delay)
print(f"Reconnecting in {actual_delay:.1f} seconds...")
time.sleep(actual_delay)
# Exponential backoff growth
self._reconnect_delay = min(self._reconnect_delay * 2, self._max_reconnect_delay)
# Reconnect with same parameters
if self.ws:
self.connect.__wrapped__(self, symbols=self._last_symbols, channels=self._last_channels)
def disconnect(self):
"""Gracefully close the WebSocket connection."""
self._running = False
if self.ws:
self.ws.close()
print("WebSocket disconnected")
# Example: Real-time dashboard handler
def handle_kline_update(data: dict):
"""Process incoming kline update."""
symbol = data.get("symbol", "UNKNOWN")
kline = data.get("kline", {})
close_price = kline.get("close", 0)
volume = kline.get("volume", 0)
timestamp = kline.get("timestamp", 0)
# Your dashboard update logic here
print(f"[{timestamp}] {symbol}: ${close_price} | Vol: {volume:,}")
# Usage
if __name__ == "__main__":
api_key = os.environ.get("TICKDB_API_KEY")
ws_client = TickDBWebSocketClient(
api_key=api_key,
on_message=handle_kline_update,
)
# Subscribe to 100 stocks on kline channel
# Single connection handles all 100 symbols — no batching required
ws_client.connect(
symbols=[
"AAPL.US", "MSFT.US", "GOOGL.US", "AMZN.US", "NVDA.US",
# ... 95 more symbols
] * 20 # Expand to 100
)
# Keep running
try:
while ws_client._running:
time.sleep(1)
except KeyboardInterrupt:
ws_client.disconnect()
Usage Monitoring and Budget Alerts
Optimization without monitoring is speculation. Implement usage tracking to catch cost anomalies before they become budget overruns.
from dataclasses import dataclass
from typing import Dict, List
import smtplib
from email.mime.text import MIMEText
@dataclass
class BudgetAlert:
threshold_calls: int
alert_email: str
current_calls: int = 0
daily_budget_usd: float = 100.0
estimated_cost_per_call: float = 0.0001
class UsageMonitor:
"""
Monitor API usage and trigger alerts before budget overruns.
Recommended configuration:
- Alert at 50% of monthly budget
- Hard stop at 80% of monthly budget (disable polling)
"""
def __init__(self, alert: BudgetAlert, client: TickDBOptimizedClient):
self.alert = alert
self.client = client
self._alert_sent = False
def check_usage(self) -> Dict:
"""Evaluate current usage against budget thresholds."""
stats = self.client.get_usage_stats()
self.alert.current_calls = stats["total_requests"]
monthly_projected = self.alert.current_calls * 30 # Assuming linear usage
estimated_cost = monthly_projected * self.alert.estimated_cost_per_call
status = {
"calls_today": self.alert.current_calls,
"monthly_projected": monthly_projected,
"estimated_cost": estimated_cost,
"budget_remaining": self.alert.daily_budget_usd - estimated_cost,
"alerts_triggered": [],
}
# Check thresholds
if estimated_cost >= self.alert.daily_budget_usd * 0.8:
status["alerts_triggered"].append("HARD_STOP_WARNING")
self._send_alert(
"URGENT: 80% Budget Threshold Reached",
f"Current spend: ${estimated_cost:.2f} / ${self.alert.daily_budget_usd:.2f}"
)
elif estimated_cost >= self.alert.daily_budget_usd * 0.5 and not self._alert_sent:
status["alerts_triggered"].append("SOFT_WARNING")
self._send_alert(
"Notice: 50% Budget Threshold Reached",
f"Consider optimizing polling frequency. "
f"Current spend: ${estimated_cost:.2f} / ${self.alert.daily_budget_usd:.2f}"
)
self._alert_sent = True
return status
def _send_alert(self, subject: str, body: str):
"""Send email alert (requires SMTP configuration)."""
# Implementation depends on your email infrastructure
print(f"[ALERT] {subject}: {body}")
Cost Optimization Summary
The table below maps each optimization technique to its impact and implementation complexity.
| Optimization | API Call Reduction | Implementation Effort | Best For |
|---|---|---|---|
| Batch requests (50/call) | 98% | Low — use /kline/latest batch endpoint |
All polling scenarios |
| Cache-first with TTL | 20–40% additional | Low — implement in-client TTL check | Dashboard apps |
| WebSocket real-time | 99%+ | Medium — async architecture needed | Sub-second requirements |
| Adaptive polling | 10–30% additional | Medium — monitor data freshness | Volatile markets |
| Symbol filtering | Variable | Low — monitor only relevant symbols | Large watchlists |
Deploying by User Segment
| User Type | Recommended Approach | Expected Monthly Cost (100 stocks, 1-min data) |
|---|---|---|
| Individual quant / researcher | Batch REST + cache-first polling | $8–$15 |
| Small team (3–5 dashboards) | Shared WebSocket connection + REST fallback | $25–$50 |
| Institutional desk | Dedicated WebSocket feeds + priority rate limits | $100–$300 |
| Enterprise backtesting | Historical kline endpoint (separate pricing) | Varies by data volume |
Closing
We began with a question that stops startups: what does this actually cost? We have answered it with formulas, code, and a path to reducing that cost by 98% without sacrificing data quality.
The discipline is not in spending less. It is in spending intentionally. Every API call should earn its place: either delivering new information or maintaining connection health. Everything else is waste.
If you are building a dashboard, start with the batch client. If you need real-time, use WebSocket. If you are doing historical research, fetch once and cache aggressively. The patterns are the same regardless of scale: batch what you can, cache what you must, and never poll what you already know.
If you want to estimate your specific use case, visit tickdb.ai and use the interactive pricing calculator with your symbol count and refresh frequency.
If you are ready to build, sign up for a free API key at tickdb.ai — no credit card required to start.
If you need historical OHLCV data for backtesting, reach out to enterprise@tickdb.ai for 10+ years of cleaned US equity data at scale.
This article does not constitute investment advice. API pricing structures are subject to change; verify current rates at tickdb.ai/pricing before building cost estimates.