"Your backtest returned 2.34 Sharpe. Your live system returned −0.8."
The gap between historical simulation and live execution rarely comes from the alpha model. More often, it comes from data governance failures — a timestamp misalignment that corrupted the training set, a provider outage that left your risk engine running on stale prices, or an audit request that your logs cannot satisfy.
For retail quant traders, these failures are expensive. For institutional quant teams managing other people's capital, they are existential. A single compliance gap can trigger regulatory sanctions, client lawsuits, or AUM redemption cascades. The Securities and Exchange Commission levied over $2.4 billion in fines related to recordkeeping violations between 2020 and 2024. Most of those violations were not fraud — they were infrastructure failures.
This article examines the three pillars of institutional data governance for quantitative systems: compliance audit readiness, disaster recovery architecture, and service-level agreement enforcement. We provide architecture patterns, implementation code, and decision frameworks for teams operating at institutional scale.
1. Why Institutional Data Governance Differs from Retail
Retail quant traders care about data accuracy and availability. Institutional teams care about the same things — plus reproducibility, auditability, and regulatory defensibility. The distinction matters because it drives architectural decisions that are expensive to retrofit.
| Dimension | Retail quant | Institutional quant |
|---|---|---|
| Data retention | Days to weeks | Years (regulatory minimums vary) |
| Audit trail | Optional log file | Immutable, tamper-evident record |
| SLA definition | "I hope the API is up" | Contractual uptime guarantees |
| Failover | Manual restart | Automatic with defined RTO/RPO |
| Regulatory exposure | None | SEC Rule 17a-4, MiFID II, Dodd-Frank |
| Change management | Code in a notebook | Version-controlled with approval workflow |
The institutional quant team operates under a fiduciary obligation to clients and a regulatory obligation to regulators. Data governance is not an engineering luxury — it is a compliance requirement.
2. Compliance Audit Requirements: What Must Be Retained
2.1 Regulatory Frameworks That Apply
The specific retention requirements depend on your jurisdiction and asset class, but the major frameworks share common themes:
SEC Rule 17a-4 (United States)
- Requires broker-dealers to retain records for six years, accessible within two years of creation.
- Covers order records, trade confirmations, account ledgers, and communications related to securities business.
- The records must be in a non-rewriteable, non-erasable format — a standard that most cloud object stores satisfy if configured correctly.
MiFID II / MiFIR (European Union)
- Requires transaction reports to regulators within 15 minutes of execution.
- Order record-keeping for five years minimum.
- Market data used in algorithm execution must be timestamped to at least millisecond precision.
Dodd-Frank Act (United States, swaps)
- Swap data repository retention for at least five years.
- Detailed audit trail of all order events, modifications, and cancellations.
2.2 The Four Categories of Institutional Records
| Category | Examples | Retention period | Format requirement |
|---|---|---|---|
| Order lifecycle | Order submission, modification, cancellation, execution | 6 years (SEC) | Immutable, machine-readable |
| Market data used in decisions | Price feeds, depth snapshots, reference data | 5–7 years | Time-synchronized, source-attributed |
| System events | Login events, API calls, configuration changes | 5 years | Tamper-evident |
| Model governance | Backtest parameters, feature definitions, performance reports | Life of fund + 5 years | Version-controlled |
2.3 TickDB Audit Trail Architecture
For teams using TickDB as a market data source, the audit trail must capture the full data consumption lifecycle — not just the data itself, but the context in which it was used.
"""
Audit logger for institutional TickDB consumption.
Logs every data request with source attribution, timestamp, and user context.
Compatible with SEC Rule 17a-4 and MiFID II recordkeeping requirements.
"""
import hashlib
import json
import logging
import os
import time
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from pathlib import Path
import threading
class InstitutionalAuditLogger:
"""
Tamper-evident audit logger for institutional market data consumption.
Features:
- Immutable append-only log format
- SHA-256 chain integrity verification
- Thread-safe for concurrent access
- Separate log streams for compliance and operational events
Note: For production SEC 17a-4 compliance, route logs to
WORM-compliant storage (e.g., AWS S3 Object Lock, Azure Immutable Blob).
"""
def __init__(
self,
audit_path: str,
chain_log_path: str,
compliance_mode: bool = True
):
"""
Initialize the institutional audit logger.
Args:
audit_path: Directory for compliance event logs
chain_log_path: Path for the integrity chain log (prev hash)
compliance_mode: If True, enforces WORM-compatible append-only writes
"""
self.audit_path = Path(audit_path)
self.chain_log_path = Path(chain_log_path)
self.compliance_mode = compliance_mode
self._lock = threading.Lock()
# Ensure directories exist with restricted permissions
self.audit_path.mkdir(parents=True, exist_ok=True)
self.chain_log_path.parent.mkdir(parents=True, exist_ok=True)
# Load previous hash for chain integrity
self._prev_hash = self._load_chain_head()
# Configure Python logging to also write to audit stream
self._setup_compliance_logging()
def _load_chain_head(self) -> str:
"""Load the last hash from the chain log for continuity."""
if self.chain_log_path.exists():
with open(self.chain_log_path, 'r') as f:
return f.read().strip()
return "GENESIS"
def _setup_compliance_logging(self):
"""Configure Python logging to write to compliance-audit stream."""
self.logger = logging.getLogger("institutional_audit")
self.logger.setLevel(logging.INFO)
# Compliance event log - JSON format for machine readability
compliance_handler = logging.FileHandler(
self.audit_path / f"compliance_{datetime.now(timezone.utc).strftime('%Y%m')}.jsonl",
mode='a' # Append-only
)
compliance_handler.setLevel(logging.INFO)
compliance_handler.setFormatter(logging.Formatter('%(message)s'))
# Remove any existing handlers to avoid duplicates
self.logger.handlers.clear()
self.logger.addHandler(compliance_handler)
def _compute_hash(self, record: Dict[str, Any]) -> str:
"""Compute SHA-256 hash including previous chain hash for tamper evidence."""
content = json.dumps(record, sort_keys=True, default=str)
chain_payload = f"{self._prev_hash}:{content}"
return hashlib.sha256(chain_payload.encode('utf-8')).hexdigest()
def log_data_request(
self,
symbol: str,
endpoint: str,
user_id: str,
request_params: Dict[str, Any],
response_status: int,
latency_ms: float,
metadata: Optional[Dict[str, Any]] = None
) -> str:
"""
Log a market data API request with full audit trail.
This method records every request to TickDB with enough context
to reconstruct the decision environment during a regulatory audit.
"""
with self._lock:
timestamp = datetime.now(timezone.utc).isoformat()
record = {
"event_type": "DATA_REQUEST",
"timestamp": timestamp,
"user_id": user_id,
"system_id": os.environ.get("SYSTEM_ID", "UNKNOWN"),
"symbol": symbol,
"endpoint": endpoint,
"request_params": request_params,
"response_status": response_status,
"latency_ms": round(latency_ms, 3),
"data_source": "TickDB",
"source_ip": os.environ.get("REQUEST_SOURCE_IP", "INTERNAL"),
"metadata": metadata or {}
}
# Compute chain-integrity hash
record["chain_hash"] = self._compute_hash(record)
record["prev_hash"] = self._prev_hash
# Emit to compliance log
self.logger.info(json.dumps(record, default=str))
# Update chain head
self._prev_hash = record["chain_hash"]
self._write_chain_head()
return record["chain_hash"]
def log_system_event(
self,
event_type: str,
description: str,
user_id: str,
severity: str = "INFO",
metadata: Optional[Dict[str, Any]] = None
) -> str:
"""
Log a system-level event (login, config change, alert).
Args:
event_type: Event category (e.g., LOGIN_SUCCESS, CONFIG_CHANGE)
description: Human-readable event description
user_id: User or service account performing the action
severity: INFO, WARNING, CRITICAL
metadata: Additional structured data
"""
with self._lock:
timestamp = datetime.now(timezone.utc).isoformat()
record = {
"event_type": event_type,
"timestamp": timestamp,
"user_id": user_id,
"system_id": os.environ.get("SYSTEM_ID", "UNKNOWN"),
"severity": severity,
"description": description,
"metadata": metadata or {}
}
record["chain_hash"] = self._compute_hash(record)
record["prev_hash"] = self._prev_hash
self.logger.info(json.dumps(record, default=str))
self._prev_hash = record["chain_hash"]
self._write_chain_head()
return record["chain_hash"]
def _write_chain_head(self):
"""Update the chain head hash. In compliance mode, this is append-only."""
with open(self.chain_log_path, 'w') as f:
f.write(self._prev_hash)
def verify_chain_integrity(self, log_path: str) -> Dict[str, Any]:
"""
Verify the integrity of a compliance log file.
Returns a report with:
- Total records verified
- Any integrity failures detected
- First and last record timestamps
"""
path = Path(log_path)
if not path.exists():
return {"status": "ERROR", "message": "Log file not found"}
prev_hash = "GENESIS"
total_records = 0
failures = []
first_timestamp = None
last_timestamp = None
with open(path, 'r') as f:
for line_num, line in enumerate(f, 1):
record = json.loads(line.strip())
total_records += 1
if first_timestamp is None:
first_timestamp = record.get("timestamp")
last_timestamp = record.get("timestamp")
# Verify chain linkage
if record.get("prev_hash") != prev_hash:
failures.append({
"line": line_num,
"expected_prev": prev_hash,
"found_prev": record.get("prev_hash"),
"issue": "Chain broken"
})
# Verify self-consistency
expected_hash = hashlib.sha256(
f"{prev_hash}:{json.dumps(record, sort_keys=True, default=str)}".encode('utf-8')
).hexdigest()
# Note: Hash includes the record itself, so we verify differently
# In production, recompute from record minus chain_hash field
record_copy = {k: v for k, v in record.items() if k not in ("chain_hash", "prev_hash")}
expected_hash = hashlib.sha256(
f"{prev_hash}:{json.dumps(record_copy, sort_keys=True, default=str)}".encode('utf-8')
).hexdigest()
if expected_hash != record.get("chain_hash"):
failures.append({
"line": line_num,
"issue": "Record hash mismatch"
})
prev_hash = record.get("chain_hash")
return {
"status": "PASS" if not failures else "FAIL",
"total_records": total_records,
"failures": failures,
"first_timestamp": first_timestamp,
"last_timestamp": last_timestamp
}
# Example usage
if __name__ == "__main__":
audit_logger = InstitutionalAuditLogger(
audit_path="/var/audit/compliance",
chain_log_path="/var/audit/.chain_head",
compliance_mode=True
)
# Log a market data request
audit_logger.log_data_request(
symbol="AAPL.US",
endpoint="/v1/market/kline",
user_id="algo_trader_01",
request_params={"interval": "1h", "limit": 100},
response_status=200,
latency_ms=42.5,
metadata={"strategy_id": "momentum_v3", "backtest_mode": False}
)
# Log a system event
audit_logger.log_system_event(
event_type="CONFIG_CHANGE",
description="Updated rate_limit_threshold from 100 to 200 req/min",
user_id="admin_jane_doe",
severity="WARNING",
metadata={"old_value": 100, "new_value": 200, "ticket_id": "CHG-2024-0892"}
)
# Verify chain integrity
report = audit_logger.verify_chain_integrity("/var/audit/compliance/compliance_2024_01.jsonl")
print(f"Integrity check: {report['status']} - {report['total_records']} records verified")
2.4 What the Audit Log Must Capture
The code above demonstrates the minimum viable audit architecture. For a defensible compliance posture, every record in your audit log should answer these five questions:
- Who: Which user account or service principal made the request?
- What: What data was requested (symbol, endpoint, parameters)?
- When: Timestamp with millisecond precision in UTC.
- Where: Source IP and system identifier.
- Why: The decision context — strategy identifier, backtest or live mode, relevant risk thresholds.
The chain-hash mechanism ensures that a regulator or auditor can detect any post-hoc tampering. If a single record is modified, its hash changes, breaking the chain.
3. Disaster Recovery Architecture
3.1 Recovery Objectives: RTO and RPO
Before designing a disaster recovery architecture, define your recovery objectives:
| Metric | Definition | Typical institutional target |
|---|---|---|
| RTO (Recovery Time Objective) | Maximum acceptable downtime | 15 minutes to 4 hours |
| RPO (Recovery Point Objective) | Maximum acceptable data loss | 0 to 15 minutes |
For a live trading system with real-time risk management, an RPO of 15 minutes means your system must have no more than 15 minutes of data loss tolerance. This drives the replication architecture.
3.2 Multi-Region Data Architecture
┌─────────────────────────────────────────────────────────────────┐
│ Primary Region (us-east-1) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ TickDB API │ │ App Server │ │ TimescaleDB │ │
│ │ Consumer │──│ (Primary) │──│ (Market Data Cache) │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
│ │ │
│ │ Async replication (write-ahead log) │
└────────────────────────────┼────────────────────────────────────┘
│
┌────────────────────────────┼────────────────────────────────────┐
│ Secondary Region (us-west-2) │
│ │ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Failover │ │ App Server │ │ TimescaleDB │ │
│ │ Load Balancer │──│ (Standby) │──│ (Read Replica) │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
│ Secondary Region (eu-west-1) │
└─────────────────────────────────────────────────────────────────┘
3.3 Hot-Warm-Cold Data Tiering
Institutional systems process multiple data types with different availability requirements. Tier your data architecture accordingly:
| Tier | Data type | Availability requirement | Typical retention | Storage |
|---|---|---|---|---|
| Hot | Current kline, depth snapshots | Sub-second access | 0–24 hours | Redis Cluster, TimescaleDB |
| Warm | Recent kline history, order flow | Minute-level access | 1–90 days | TimescaleDB, S3 |
| Cold | Historical backtest data | Hour-level access | 7+ years | S3 Glacier, Iceberg |
| Archive | Audit logs, compliance records | On-demand retrieval | Regulatory minimum | WORM storage |
3.4 Multi-Provider Fallback Implementation
For a robust data ingestion pipeline, implement automatic fallback across providers. The following code demonstrates a TickDB-first architecture with automatic failover to a secondary provider:
"""
Multi-provider market data ingestion with automatic failover.
Implements hot-standby architecture for institutional DR requirements.
"""
import os
import time
import threading
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Callable, Dict, List, Optional, Tuple
import requests
logger = logging.getLogger("market_data_ingestion")
class ProviderStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
FAILED = "failed"
@dataclass
class ProviderConfig:
"""Configuration for a market data provider."""
name: str
base_url: str
api_key_env_var: str
priority: int = 0 # Lower = higher priority
timeout_seconds: float = 5.0
health_check_endpoint: str = "/v1/system/health"
max_retries: int = 3
backoff_base_seconds: float = 1.0
@dataclass
class HealthMetrics:
"""Real-time health metrics for a provider."""
provider_name: str
latency_p50_ms: float = 0.0
latency_p99_ms: float = 0.0
error_rate: float = 0.0
last_success: Optional[datetime] = None
consecutive_failures: int = 0
status: ProviderStatus = ProviderStatus.HEALTHY
class MultiProviderIngestionService:
"""
Multi-provider market data ingestion with automatic failover.
Features:
- Active-passive provider architecture (TickDB primary)
- Real-time health monitoring with adaptive routing
- Circuit breaker pattern to isolate failing providers
- Consistent data normalization across providers
Usage:
service = MultiProviderIngestionService()
service.register_provider(tickdb_config)
service.register_provider(backup_config)
service.start()
# Get data - automatically routes to best available provider
data = service.get_kline("AAPL.US", "1h", limit=100)
"""
def __init__(self, circuit_breaker_threshold: int = 5):
"""
Initialize the multi-provider ingestion service.
Args:
circuit_breaker_threshold: Number of consecutive failures
before a provider is isolated
"""
self._providers: Dict[str, ProviderConfig] = {}
self._health_metrics: Dict[str, HealthMetrics] = {}
self._lock = threading.RLock()
self._circuit_breakers: Dict[str, int] = {}
self._circuit_breaker_threshold = circuit_breaker_threshold
self._running = False
self._health_check_interval = 30 # seconds
# Fallback data cache for RPO compliance
self._fallback_cache: Dict[str, Tuple[any, datetime]] = {}
self._cache_max_age_seconds = 300 # 5 minutes
self.logger = logging.getLogger("market_data_ingestion")
def register_provider(self, config: ProviderConfig):
"""Register a market data provider with the ingestion service."""
with self._lock:
self._providers[config.name] = config
self._health_metrics[config.name] = HealthMetrics(
provider_name=config.name
)
self._circuit_breakers[config.name] = 0
self.logger.info(
f"Registered provider: {config.name} "
f"(priority={config.priority}, base_url={config.base_url})"
)
def start(self):
"""Start the background health monitoring thread."""
self._running = True
self._health_thread = threading.Thread(
target=self._health_monitor_loop,
daemon=True,
name="HealthMonitor"
)
self._health_thread.start()
self.logger.info("Multi-provider ingestion service started")
def stop(self):
"""Stop the ingestion service and health monitoring."""
self._running = False
self.logger.info("Multi-provider ingestion service stopped")
def _health_monitor_loop(self):
"""Background thread that monitors provider health."""
while self._running:
self._check_all_providers()
time.sleep(self._health_check_interval)
def _check_all_providers(self):
"""Check health of all registered providers."""
for name, config in self._providers.items():
try:
start = time.perf_counter()
response = requests.get(
f"{config.base_url}{config.health_check_endpoint}",
headers={"X-API-Key": os.environ.get(config.api_key_env_var)},
timeout=config.timeout_seconds
)
latency_ms = (time.perf_counter() - start) * 1000
with self._lock:
metrics = self._health_metrics[name]
metrics.last_success = datetime.now(timezone.utc)
metrics.consecutive_failures = 0
# Update latency percentiles (simplified EMA)
metrics.latency_p50_ms = (
0.9 * metrics.latency_p50_ms + 0.1 * latency_ms
)
if response.status_code == 200:
metrics.status = ProviderStatus.HEALTHY
self._circuit_breakers[name] = 0
else:
metrics.status = ProviderStatus.DEGRADED
self.logger.debug(
f"Health check {name}: OK (latency={latency_ms:.1f}ms)"
)
except Exception as e:
with self._lock:
metrics = self._health_metrics[name]
metrics.consecutive_failures += 1
metrics.error_rate = min(
metrics.error_rate + 0.1, 1.0
)
if metrics.consecutive_failures >= self._circuit_breaker_threshold:
metrics.status = ProviderStatus.FAILED
self.logger.warning(
f"Provider {name} circuit breaker OPEN "
f"({metrics.consecutive_failures} consecutive failures)"
)
self.logger.error(f"Health check {name}: FAILED - {e}")
def _get_available_provider(self) -> Optional[Tuple[ProviderConfig, HealthMetrics]]:
"""
Select the best available provider using priority and health metrics.
Returns:
Tuple of (provider_config, health_metrics) for the selected provider,
or None if no providers are available.
"""
with self._lock:
available = []
for name, config in self._providers.items():
metrics = self._health_metrics[name]
# Skip providers with open circuit breakers
if self._circuit_breakers.get(name, 0) >= self._circuit_breaker_threshold:
continue
# Skip providers in FAILED status
if metrics.status == ProviderStatus.FAILED:
continue
# Score based on priority and health
score = (
(10 - config.priority) * 100 + # Priority weight
(1 - metrics.error_rate) * 50 + # Error rate weight
(1 / (1 + metrics.latency_p50_ms / 1000)) * 50 # Latency weight
)
available.append((score, config, metrics))
if not available:
return None
# Sort by score descending and return the best
available.sort(key=lambda x: x[0], reverse=True)
return available[0][1], available[0][2]
def _execute_with_fallback(
self,
request_func: Callable[[ProviderConfig], requests.Response]
) -> Optional[requests.Response]:
"""
Execute a request with automatic fallback to secondary providers.
Args:
request_func: Function that executes the request given a provider config
Returns:
Response object, or None if all providers failed
"""
tried_providers = []
# Sort providers by priority
with self._lock:
sorted_providers = sorted(
self._providers.items(),
key=lambda x: x[1].priority
)
for name, config in sorted_providers:
try:
response = request_func(config)
if response.status_code in (200, 429): # 429 = rate limited, not failed
return response
tried_providers.append((name, response.status_code))
except Exception as e:
with self._lock:
metrics = self._health_metrics[name]
metrics.consecutive_failures += 1
if metrics.consecutive_failures >= self._circuit_breaker_threshold:
self._circuit_breakers[name] = self._circuit_breaker_threshold
metrics.status = ProviderStatus.FAILED
tried_providers.append((name, str(e)))
continue
self.logger.error(
f"All providers failed. Tried: {tried_providers}"
)
return None
def get_kline(
self,
symbol: str,
interval: str,
limit: int = 100,
start_time: Optional[int] = None,
end_time: Optional[int] = None
) -> Optional[Dict]:
"""
Fetch kline (OHLCV) data with multi-provider fallback.
Args:
symbol: Trading symbol (e.g., "AAPL.US")
interval: Candle interval (e.g., "1m", "1h", "1d")
limit: Maximum number of candles to return
start_time: Start timestamp in milliseconds (optional)
end_time: End timestamp in milliseconds (optional)
Returns:
Dictionary with kline data, or None if all providers failed
"""
def request(config: ProviderConfig) -> requests.Response:
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
if start_time:
params["start"] = start_time
if end_time:
params["end"] = end_time
return requests.get(
f"{config.base_url}/v1/market/kline",
headers={"X-API-Key": os.environ.get(config.api_key_env_var)},
params=params,
timeout=config.timeout_seconds
)
response = self._execute_with_fallback(request)
if response is None:
# Return cached data if available (for RPO compliance)
return self._get_from_cache(symbol, interval)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
self.logger.warning(f"Rate limited. Retrying after {retry_after}s")
time.sleep(retry_after)
return self.get_kline(symbol, interval, limit, start_time, end_time)
data = response.json()
if data.get("code") == 0:
# Cache successful response
self._put_in_cache(symbol, interval, data)
return data.get("data")
return None
def _get_from_cache(self, symbol: str, interval: str) -> Optional[Dict]:
"""Retrieve data from fallback cache if within age limit."""
cache_key = f"{symbol}:{interval}"
with self._lock:
if cache_key in self._fallback_cache:
data, timestamp = self._fallback_cache[cache_key]
age_seconds = (datetime.now(timezone.utc) - timestamp).total_seconds()
if age_seconds < self._cache_max_age_seconds:
self.logger.warning(
f"Serving stale cache data for {symbol} "
f"(age={age_seconds:.0f}s, RPO={self._cache_max_age_seconds}s)"
)
return data
else:
del self._fallback_cache[cache_key]
return None
def _put_in_cache(self, symbol: str, interval: str, data: Dict):
"""Store data in fallback cache for RPO compliance."""
cache_key = f"{symbol}:{interval}"
with self._lock:
self._fallback_cache[cache_key] = (
data,
datetime.now(timezone.utc)
)
def get_health_report(self) -> Dict[str, HealthMetrics]:
"""Return the current health status of all providers."""
with self._lock:
return dict(self._health_metrics)
# Example configuration for institutional deployment
if __name__ == "__main__":
# Initialize service
service = MultiProviderIngestionService(
circuit_breaker_threshold=5
)
# Register TickDB as primary provider
service.register_provider(ProviderConfig(
name="tickdb_primary",
base_url="https://api.tickdb.ai",
api_key_env_var="TICKDB_API_KEY",
priority=1, # Highest priority
timeout_seconds=5.0
))
# Register backup provider
service.register_provider(ProviderConfig(
name="tickdb_secondary",
base_url="https://backup-api.tickdb.ai",
api_key_env_var="TICKDB_API_KEY_BACKUP",
priority=2, # Fallback
timeout_seconds=8.0
))
# Start the ingestion service
service.start()
# Fetch data - automatically uses best available provider
try:
data = service.get_kline("AAPL.US", "1h", limit=100)
print(f"Retrieved {len(data.get('klines', []))} candles for AAPL.US")
finally:
service.stop()
3.5 Recovery Playbook
Document and test your recovery procedures before you need them:
| Scenario | RTO target | Recovery steps |
|---|---|---|
| Primary API region outage | 5 minutes | Failover to secondary region via DNS update |
| Database corruption | 30 minutes | Promote read replica; replay from WAL |
| Application crash | 2 minutes | Auto-restart via systemd; health check before traffic |
| Data provider failure | 0 minutes (automatic) | Circuit breaker opens; fallback provider activates |
| Region-wide failure | 15 minutes | Failover to cross-region deployment |
Test your recovery procedures quarterly with chaos engineering principles — deliberately introduce failures in a staging environment to verify your detection, isolation, and recovery pipelines.
4. SLA Enforcement and Monitoring
4.1 Defining SLAs for Market Data Infrastructure
Service Level Agreements for institutional quant systems should cover three dimensions:
| SLA dimension | Metric | Typical target | Measurement method |
|---|---|---|---|
| Availability | Uptime percentage | 99.9% (8.76 hours downtime/year) | Synthetic monitoring + real user monitoring |
| Latency | P95 and P99 response time | P95 < 100ms, P99 < 500ms | APM agent, distributed tracing |
| Data completeness | Successful data points / total requested | > 99.95% | Reconciliation against expected data volume |
4.2 SLA Monitoring Implementation
"""
SLA monitoring dashboard and alerting for institutional quant systems.
Integrates with PagerDuty, Slack, and enterprise SIEM platforms.
"""
import os
import time
import logging
import threading
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Dict, List, Optional
from collections import deque
import statistics
logger = logging.getLogger("sla_monitor")
class AlertSeverity(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class SLAMetric:
"""A single SLA metric measurement."""
timestamp: datetime
metric_name: str
value: float
unit: str
target: float
provider: str
@dataclass
class SLAReport:
"""Aggregated SLA report for a time window."""
window_start: datetime
window_end: datetime
total_requests: int
successful_requests: int
failed_requests: int
availability_pct: float
latency_p50_ms: float
latency_p95_ms: float
latency_p99_ms: float
data_completeness_pct: float
breaches: List[Dict]
def to_dict(self) -> Dict:
return {
"window": f"{self.window_start.isoformat()} to {self.window_end.isoformat()}",
"availability": f"{self.availability_pct:.3f}%",
"latency_p50": f"{self.latency_p50_ms:.1f}ms",
"latency_p95": f"{self.latency_p95_ms:.1f}ms",
"latency_p99": f"{self.latency_p99_ms:.1f}ms",
"data_completeness": f"{self.data_completeness_pct:.3f}%",
"total_requests": self.total_requests,
"failures": self.failed_requests,
"breaches": self.breaches
}
class SLAMonitor:
"""
Real-time SLA monitoring for market data infrastructure.
Tracks availability, latency, and data completeness metrics.
Generates alerts when SLAs are at risk of breach.
Usage:
monitor = SLAMonitor(
availability_target=99.9,
latency_p95_target_ms=200,
reporting_interval_seconds=300
)
monitor.start()
# Record measurements
monitor.record_request(endpoint="/v1/market/kline", latency_ms=45.2, success=True)
monitor.record_data_point(symbol="AAPL.US", received=True)
"""
def __init__(
self,
availability_target: float = 99.9,
latency_p95_target_ms: float = 200.0,
latency_p99_target_ms: float = 500.0,
data_completeness_target: float = 99.95,
reporting_interval_seconds: int = 300,
alert_webhook_url: Optional[str] = None
):
self.availability_target = availability_target
self.latency_p95_target_ms = latency_p95_target_ms
self.latency_p99_target_ms = latency_p99_target_ms
self.data_completeness_target = data_completeness_target
self.reporting_interval = reporting_interval_seconds
self.alert_webhook_url = alert_webhook_url
# Sliding window for metrics (last hour, 1-minute granularity)
self._request_latencies: deque = deque(maxlen=3600)
self._request_results: deque = deque(maxlen=3600)
self._data_points: deque = deque(maxlen=3600)
self._lock = threading.Lock()
self._running = False
self._last_report_time = datetime.now(timezone.utc)
self.logger = logging.getLogger("sla_monitor")
def start(self):
"""Start the SLA monitoring background thread."""
self._running = True
self._monitor_thread = threading.Thread(
target=self._monitoring_loop,
daemon=True,
name="SLAMonitor"
)
self._monitor_thread.start()
self.logger.info("SLA monitor started")
def stop(self):
"""Stop the SLA monitoring thread."""
self._running = False
self.logger.info("SLA monitor stopped")
def record_request(
self,
endpoint: str,
latency_ms: float,
success: bool,
provider: str = "tickdb"
):
"""Record a single API request for SLA tracking."""
with self._lock:
self._request_latencies.append({
"timestamp": datetime.now(timezone.utc),
"endpoint": endpoint,
"latency_ms": latency_ms,
"provider": provider
})
self._request_results.append({
"timestamp": datetime.now(timezone.utc),
"success": success,
"provider": provider
})
def record_data_point(
self,
symbol: str,
interval: str,
received: bool,
provider: str = "tickdb"
):
"""Record a data completeness measurement."""
with self._lock:
self._data_points.append({
"timestamp": datetime.now(timezone.utc),
"symbol": symbol,
"interval": interval,
"received": received,
"provider": provider
})
def _monitoring_loop(self):
"""Background thread that generates periodic reports and alerts."""
while self._running:
time.sleep(self.reporting_interval)
self._generate_report_and_alert()
def _generate_report_and_alert(self):
"""Generate SLA report and trigger alerts if thresholds are breached."""
now = datetime.now(timezone.utc)
window_start = self._last_report_time
self._last_report_time = now
with self._lock:
# Filter metrics within the window
request_latencies = [r for r in self._request_latencies
if window_start <= r["timestamp"] <= now]
request_results = [r for r in self._request_results
if window_start <= r["timestamp"] <= now]
data_points = [d for d in self._data_points
if window_start <= d["timestamp"] <= now]
if not request_results:
return
# Calculate metrics
total_requests = len(request_results)
successful_requests = sum(1 for r in request_results if r["success"])
failed_requests = total_requests - successful_requests
availability_pct = (successful_requests / total_requests) * 100
latencies = [r["latency_ms"] for r in request_latencies]
latency_p50 = statistics.median(latencies) if latencies else 0
latency_p95 = self._percentile(latencies, 0.95) if latencies else 0
latency_p99 = self._percentile(latencies, 0.99) if latencies else 0
total_data_points = len(data_points)
received_data_points = sum(1 for d in data_points if d["received"])
data_completeness_pct = (
(received_data_points / total_data_points * 100)
if total_data_points > 0 else 100.0
)
# Identify breaches
breaches = []
if availability_pct < self.availability_target:
breaches.append({
"metric": "availability",
"actual": f"{availability_pct:.3f}%",
"target": f"{self.availability_target}%",
"severity": "CRITICAL" if availability_pct < 99.0 else "WARNING"
})
if latency_p95 > self.latency_p95_target_ms:
breaches.append({
"metric": "latency_p95",
"actual": f"{latency_p95:.1f}ms",
"target": f"{self.latency_p95_target_ms}ms",
"severity": "WARNING"
})
if latency_p99 > self.latency_p99_target_ms:
breaches.append({
"metric": "latency_p99",
"actual": f"{latency_p99:.1f}ms",
"target": f"{self.latency_p99_target_ms}ms",
"severity": "WARNING"
})
if data_completeness_pct < self.data_completeness_target:
breaches.append({
"metric": "data_completeness",
"actual": f"{data_completeness_pct:.3f}%",
"target": f"{self.data_completeness_target}%",
"severity": "CRITICAL" if data_completeness_pct < 99.0 else "WARNING"
})
report = SLAReport(
window_start=window_start,
window_end=now,
total_requests=total_requests,
successful_requests=successful_requests,
failed_requests=failed_requests,
availability_pct=availability_pct,
latency_p50_ms=latency_p50,
latency_p95_ms=latency_p95,
latency_p99_ms=latency_p99,
data_completeness_pct=data_completeness_pct,
breaches=breaches
)
# Log report
self.logger.info(f"SLA Report: {report.to_dict()}")
# Trigger alerts for breaches
if breaches:
self._send_alerts(report)
return report
def _percentile(self, data: List[float], p: float) -> float:
"""Calculate the p-th percentile of a list."""
if not data:
return 0.0
sorted_data = sorted(data)
index = int(len(sorted_data) * p)
return sorted_data[min(index, len(sorted_data) - 1)]
def _send_alerts(self, report: SLAReport):
"""Send alerts to configured notification endpoints."""
critical_breaches = [b for b in report.breaches if b["severity"] == "CRITICAL"]
if not critical_breaches and not self.alert_webhook_url:
return
alert_payload = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"report": report.to_dict(),
"breaches": report.breaches,
"severity": "CRITICAL" if critical_breaches else "WARNING"
}
# Log critical breaches immediately
if critical_breaches:
self.logger.critical(
f"SLA BREACH: {len(critical_breaches)} critical breach(es) detected. "
f"Availability: {report.availability_pct:.3f}%, "
f"Data completeness: {report.data_completeness_pct:.3f}%"
)
# Send to webhook if configured
if self.alert_webhook_url:
try:
import requests
requests.post(
self.alert_webhook_url,
json=alert_payload,
timeout=5
)
except Exception as e:
self.logger.error(f"Failed to send alert webhook: {e}")
def get_current_metrics(self) -> Dict:
"""Return current SLA metrics (last reporting window)."""
with self._lock:
return {
"pending_requests": len(self._request_results),
"pending_data_points": len(self._data_points),
"availability_target": self.availability_target,
"latency_p95_target_ms": self.latency_p95_target_ms
}
# Example usage
if __name__ == "__main__":
monitor = SLAMonitor(
availability_target=99.9,
latency_p95_target_ms=200.0,
data_completeness_target=99.95,
reporting_interval_seconds=60,
alert_webhook_url=os.environ.get("ALERT_WEBHOOK_URL")
)
monitor.start()
# Simulate incoming requests
import random
for i in range(100):
monitor.record_request(
endpoint="/v1/market/kline",
latency_ms=random.gauss(50, 20),
success=random.random() > 0.02 # 98% success rate
)
monitor.record_data_point(
symbol="AAPL.US",
interval="1h",
received=random.random() > 0.01 # 99% completeness
)
time.sleep(0.1)
# Get current status
status = monitor.get_current_metrics()
print(f"Current metrics: {status}")
monitor.stop()
5. Data Governance Decision Framework
Institutional quant teams face resource constraints. Not every governance initiative can be pursued simultaneously. Use this framework to prioritize:
| Priority | Initiative | Why first | Estimated effort |
|---|---|---|---|
| P0 | Audit log with chain integrity | Regulatory exposure is immediate and existential | 1–2 weeks |
| P0 | Data provider fallback | A single API outage can halt trading and trigger risk limits | 1 week |
| P1 | SLA monitoring | Required for vendor contract enforcement and client reporting | 1 week |
| P1 | Multi-region deployment | Required for RTO < 4 hours | 4–8 weeks |
| P2 | Full DR automation | Valuable but expensive; evaluate after P0/P1 | 8–12 weeks |
6. Conclusion
Institutional data governance is not a checklist — it is an operational philosophy. The systems you build to ensure compliance, withstand failures, and guarantee service levels are the same systems that protect your Sharpe ratio from degrading into negative territory.
The three pillars — audit trails, disaster recovery, and SLA monitoring — are deeply interconnected. Your audit log proves that you made the right decision. Your DR architecture ensures you had the data to make any decision at all. Your SLA monitoring tells you when your data provider has broken its promise before it becomes your problem.
Start with the audit log. Chain-hash every request. Make it immutable. Test the integrity verification quarterly. Then build outward from there.
The difference between a quant fund that survives a regulatory audit and one that does not is rarely about the alpha model. It is about whether you can prove, byte by byte, that your data was clean, your decisions were logged, and your systems were available when they needed to be.
Next Steps
If you're building an institutional quant system, ensure your data infrastructure meets audit and DR requirements before you deploy capital. Contact enterprise@tickdb.ai for data governance consultation and multi-region deployment options.
If you need to verify your current data source's compliance posture, review TickDB's SLA documentation and uptime history at tickdb.ai before your next regulatory review.
If you're an institutional quant team evaluating data vendors, request the SOC 2 Type II report and data residency documentation from your provider. These are table stakes for institutional onboarding.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Data governance requirements vary by jurisdiction and asset class. Consult your compliance team for regulatory guidance specific to your organization.