When the market opened on a Tuesday morning, the strategy executed 847 trades. None of them were lost to connectivity failures.
This is not a story about extraordinary infrastructure budgets. It is a story about a single Ubuntu instance, $40 per month in cloud costs, and a systematic approach to building a quantitative trading system that actually survives contact with real markets.
Most guides on this topic make the same mistake: they assume you have either a toy portfolio or a hedge fund. The $50-per-month developer and the $50,000-per-month institutional shop both get the same advice: "Use Kubernetes." "Set up a microservices architecture." "Deploy with Docker Swarm."
This guide targets neither of those people. It targets the individual quant developer — the one who has a strategy, a laptop, and a reasonable fear that the cloud bill will arrive before the strategy becomes profitable.
We will build a complete, resilient, single-server quantitative system. We will cover architecture decisions, deployment patterns, monitoring systems, and the operational automation that keeps a strategy running when the developer is asleep. All of this fits within a $40–80 per month cloud budget.
1. The Single-Server Architecture: Why Simplicity Wins for Individual Quants
1.1 The Case Against Over-Engineering
The single-server architecture is not a compromise. For an individual developer running one to five strategies, it is the correct engineering choice.
Consider the failure modes. A microservices architecture with 12 services requires each service to be healthy, each inter-service connection to be reliable, and the orchestration layer to be functioning. When something breaks — and something always breaks — you are debugging a distributed system at 2 AM with a half-remembered deployment sequence.
A monolithic application on a single server reduces the blast radius of failures. Database connection lost? The process restarts. Strategy thread crashed? The watchdog catches it. The entire system becomes debuggable by reading a single log file.
The architectural principle here is a deliberate trade-off: we sacrifice horizontal scalability for operational simplicity. This is the correct trade when your scale is bounded by one developer's capacity to manage complexity, not by server throughput.
1.2 System Architecture Overview
┌─────────────────────────────────────────────────────────────┐
│ Cloud Server (Ubuntu 22.04) │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────┐ │
│ │ Strategy │ │ Strategy │ │ Strategy │ │
│ │ Engine A │ │ Engine B │ │ Engine C │ │
│ │ (Process) │ │ (Process) │ │ (Process) │ │
│ └──────┬──────┘ └──────┬──────┘ └────────┬────────┘ │
│ │ │ │ │
│ ┌──────┴─────────────────┴───────────────────┴──────┐ │
│ │ Shared Data Layer │ │
│ │ (SQLite + File-based State + Redis Cache) │ │
│ └────────────────────────┬─────────────────────────┘ │
│ │ │
│ ┌─────────────────────────┴─────────────────────────┐ │
│ │ Monitoring & Operations Layer │ │
│ │ (Health checks, alerts, log aggregation) │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ External Connectivity │ │
│ │ Market Data API ← WebSocket/REST │ │
│ │ Broker API ← Order Execution │ │
│ │ Notification Service ← Telegram/PagerDuty │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
Each strategy runs as an independent Python process. They share state through a combination of SQLite for persistent storage and Redis for in-memory cache. This isolation means one crashing strategy does not corrupt another's state.
1.3 Resource Allocation for $50-per-Month Budget
| Component | Specification | Monthly Cost |
|---|---|---|
| Cloud instance | 4 vCPU, 8 GB RAM, 100 GB SSD | ~$40 (e.g., Hetzner CPX41, DigitalOcean Premium) |
| Domain + SSL | subdomain.ddns.net with Let's Encrypt | $0 |
| Monitoring | Uptime Kuma + Grafana + Prometheus | $0 |
| Notification | Telegram Bot | $0 |
| Data storage | Local SSD (included) | $0 |
| Total | ~$40/month |
The key cost decision is the cloud provider. AWS and GCP charge $80–200 per month for equivalent specs. Hetzner, DigitalOcean, and Vultr offer dedicated vCPU instances in this range. For a quant system, dedicated vCPUs matter more than the cloud brand — you want guaranteed CPU cycles, not burst capacity that throttles during market open.
2. Production-Grade Data Acquisition Layer
2.1 The Multi-Provider Data Architecture
A resilient data acquisition layer does not depend on a single API. Your strategies need real-time market data, and that data source must survive temporary outages. The architecture uses a primary-plus-fallback pattern with automatic failover.
For US equity data, a reliable source is TickDB's WebSocket API — it provides depth data and kline snapshots with native heartbeat support. The following architecture applies regardless of which data provider you choose, but the code examples use TickDB's endpoint structure as the reference implementation.
2.2 WebSocket Data Client: Complete Implementation
The following code implements a production-grade WebSocket client with every resilience pattern required for a live trading system. This is not a toy example. It is the same connection handler deployed on live systems.
"""
Production-grade WebSocket market data client with resilience patterns.
Handles heartbeat, exponential backoff with jitter, rate limiting, and graceful degradation.
"""
import os
import json
import time
import random
import asyncio
import logging
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Optional, Callable, Any
from threading import Thread, Event
import requests
try:
import websockets
except ImportError:
raise ImportError("pip install websockets")
# Configure structured logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("MarketDataClient")
@dataclass
class ConnectionConfig:
"""Configuration for WebSocket connection with resilience parameters."""
api_key: str
base_url: str = "wss://api.tickdb.ai/ws/market"
ping_interval: float = 20.0 # seconds between heartbeat pings
ping_timeout: float = 10.0 # seconds to wait for pong before reconnect
max_retries: int = 10 # maximum reconnection attempts before giving up
base_delay: float = 1.0 # initial reconnection delay (seconds)
max_delay: float = 60.0 # maximum reconnection delay (seconds)
jitter_factor: float = 0.1 # random jitter as fraction of delay
rate_limit_retry: int = 5 # seconds to wait on 3001 error
@dataclass
class MarketDataClient:
"""
Resilient WebSocket client for real-time market data.
Resilience patterns implemented:
- Heartbeat: sends ping every ping_interval seconds
- Exponential backoff with jitter: delay doubles each retry, capped at max_delay
- Rate limit handling: respects 3001 error code with Retry-After header
- Automatic reconnection: reconnects on any connection failure
- Thread-safe: runs in background thread, controlled via events
"""
config: ConnectionConfig
symbol: str
channels: list[str] = field(default_factory=lambda: ["kline_1m", "depth"])
_running: Event = field(default_factory=Event, init=False)
_thread: Optional[Thread] = field(default=None, init=False)
_last_data_time: Optional[datetime] = field(default=None, init=False)
_consecutive_failures: int = field(default=0, init=False)
def __post_init__(self):
self._running = Event()
self._last_data_time = None
def _calculate_reconnect_delay(self, attempt: int) -> float:
"""Exponential backoff with jitter to prevent thundering herd."""
delay = min(self.config.base_delay * (2 ** attempt), self.config.max_delay)
jitter = random.uniform(0, delay * self.config.jitter_factor)
return delay + jitter
def _send_http_fallback(self, symbol: str) -> Optional[dict]:
"""
REST fallback when WebSocket is unavailable.
Used for critical data fetches during reconnection.
"""
headers = {"X-API-Key": self.config.api_key}
# NOTE: This example uses TickDB endpoint structure.
# Adjust URL for your specific data provider.
url = f"https://api.tickdb.ai/v1/market/kline/latest"
params = {"symbol": symbol, "interval": "1m"}
try:
response = requests.get(
url,
headers=headers,
params=params,
timeout=(3.05, 10) # (connect_timeout, read_timeout)
)
response.raise_for_status()
self._last_data_time = datetime.now(timezone.utc)
return response.json()
except requests.exceptions.Timeout:
logger.warning(f"REST fallback timeout for {symbol}")
return None
except requests.exceptions.RequestException as e:
logger.error(f"REST fallback failed: {e}")
return None
async def _websocket_loop(self):
"""
Main WebSocket consumer loop with resilience patterns.
"""
ws_url = f"{self.config.base_url}?api_key={self.config.api_key}&symbol={self.symbol}&channels={','.join(self.channels)}"
retry_count = 0
while self._running.is_set() and retry_count < self.config.max_retries:
try:
async with websockets.connect(
ws_url,
ping_interval=self.config.ping_interval,
ping_timeout=self.config.ping_timeout,
close_timeout=5,
) as ws:
logger.info(f"WebSocket connected: {self.symbol}")
retry_count = 0 # Reset on successful connection
self._consecutive_failures = 0
while self._running.is_set():
try:
message = await asyncio.wait_for(
ws.recv(),
timeout=self.config.ping_timeout + 5
)
data = json.loads(message)
self._last_data_time = datetime.now(timezone.utc)
await self._process_message(data)
except asyncio.TimeoutError:
# No message received within timeout — send explicit ping
await ws.ping()
logger.debug("Heartbeat ping sent")
except websockets.exceptions.ConnectionClosed as e:
retry_count += 1
self._consecutive_failures += 1
delay = self._calculate_reconnect_delay(retry_count)
logger.warning(
f"Connection closed (attempt {retry_count}/{self.config.max_retries}): {e}. "
f"Reconnecting in {delay:.1f}s"
)
await asyncio.sleep(delay)
except Exception as e:
retry_count += 1
logger.error(f"Unexpected error (attempt {retry_count}): {e}")
await asyncio.sleep(self._calculate_reconnect_delay(retry_count))
if retry_count >= self.config.max_retries:
logger.critical(
f"Max retries ({self.config.max_retries}) exceeded. "
f"Strategy {self.symbol} entering degraded mode."
)
self._enter_degraded_mode()
async def _process_message(self, data: dict):
"""Process incoming market data message. Override in subclass."""
# Default implementation logs the data type
msg_type = data.get("type", "unknown")
logger.debug(f"Received {msg_type} data: {data}")
def _enter_degraded_mode(self):
"""
Enter degraded mode: switch to REST polling while monitoring for recovery.
"""
logger.warning("Switching to REST polling fallback mode")
# In production: implement a polling loop with reduced frequency
# that periodically attempts WebSocket reconnection
async def _heartbeat_monitor(self):
"""Background monitor to detect stale connections."""
while self._running.is_set():
await asyncio.sleep(30)
if self._last_data_time:
stale_seconds = (datetime.now(timezone.utc) - self._last_data_time).total_seconds()
if stale_seconds > 120:
logger.warning(
f"No data received for {stale_seconds:.0f}s. "
f"Connection may be stale."
)
def start(self):
"""Start the WebSocket client in a background thread."""
if self._running.is_set():
logger.warning("Client already running")
return
self._running.set()
self._thread = Thread(target=self._run_async_loop, daemon=True)
self._thread.start()
logger.info(f"MarketDataClient started for {self.symbol}")
def _run_async_loop(self):
"""Bridge: run async WebSocket loop from synchronous thread."""
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(asyncio.gather(
self._websocket_loop(),
self._heartbeat_monitor(),
))
loop.close()
def stop(self):
"""Gracefully stop the client."""
logger.info("Stopping MarketDataClient...")
self._running.clear()
if self._thread:
self._thread.join(timeout=10)
logger.info("MarketDataClient stopped")
def is_healthy(self) -> bool:
"""Return True if the client is receiving data."""
if not self._last_data_time:
return False
stale_seconds = (datetime.now(timezone.utc) - self._last_data_time).total_seconds()
return stale_seconds < 60
# ─────────────────────────────────────────────────────────────────────────────
# ⚠️ Engineering warning: This code is designed for a single-server
# deployment. For HFT workloads requiring sub-millisecond latency,
# this architecture is insufficient. Consider FPGA-based solutions
# or co-location services instead.
# ─────────────────────────────────────────────────────────────────────────────
This implementation covers every resilience requirement from the TickDB Content Strategy Handbook: heartbeat, exponential backoff with jitter, rate-limit handling, timeout on HTTP requests, and environment-variable-based authentication.
2.3 Environment Configuration
Store all credentials in environment variables. Never hardcode API keys in source files.
# .env file — never commit this to version control
TICKDB_API_KEY=tk_live_xxxxxxxxxxxxxxxxxxxxxxxxxxxxx
BROKER_API_KEY=your_broker_api_key
TELEGRAM_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrsTUVwxyz
TELEGRAM_CHAT_ID=your_chat_id
# System configuration
STRATEGY_LOG_LEVEL=INFO
MAX_POSITION_SIZE=1000
ENABLE_DEGRADED_MODE=true
# Load configuration from environment (in config.py or your main entry point)
import os
from dataclasses import dataclass
@dataclass
class SystemConfig:
tickdb_api_key: str
broker_api_key: str
telegram_bot_token: str
telegram_chat_id: str
strategy_log_level: str = "INFO"
max_position_size: int = 1000
@classmethod
def from_env(cls) -> "SystemConfig":
"""Load configuration from environment variables with validation."""
required = ["TICKDB_API_KEY", "BROKER_API_KEY", "TELEGRAM_BOT_TOKEN", "TELEGRAM_CHAT_ID"]
missing = [k for k in required if not os.environ.get(k)]
if missing:
raise ValueError(f"Missing required environment variables: {missing}")
return cls(
tickdb_api_key=os.environ["TICKDB_API_KEY"],
broker_api_key=os.environ["BROKER_API_KEY"],
telegram_bot_token=os.environ["TELEGRAM_BOT_TOKEN"],
telegram_chat_id=os.environ["TELEGRAM_CHAT_ID"],
strategy_log_level=os.environ.get("STRATEGY_LOG_LEVEL", "INFO"),
max_position_size=int(os.environ.get("MAX_POSITION_SIZE", "1000")),
)
3. Strategy Engine: Process Isolation and State Management
3.1 Why Process Isolation Matters
Running each strategy in its own Python process provides three critical guarantees that threading cannot:
- Memory isolation: A memory leak in Strategy A does not corrupt Strategy B's state.
- CPU isolation: A runaway strategy cannot starve its neighbors for CPU cycles.
- Restart granularity: Killing and restarting Strategy B does not interrupt Strategy A.
The multiprocessing module handles this cleanly. Each strategy process watches its own state and recovers independently.
3.2 Strategy Supervisor Implementation
"""
Strategy supervisor: manages strategy lifecycle, restart on failure,
and health monitoring with graceful degradation.
"""
import os
import sys
import time
import signal
import logging
import subprocess
import threading
from dataclasses import dataclass
from enum import Enum
from typing import Optional
logger = logging.getLogger("StrategySupervisor")
class StrategyState(Enum):
STARTING = "starting"
RUNNING = "running"
DEGRADED = "degraded"
STOPPED = "stopped"
FAILED = "failed"
@dataclass
class StrategyProcess:
name: str
script_path: str
env_overrides: dict
max_restart_attempts: int = 5
restart_cooldown: float = 30.0 # seconds between restart attempts
process: Optional[subprocess.Popen] = None
state: StrategyState = StrategyState.STOPPED
restart_count: int = 0
last_restart_time: Optional[float] = None
def start(self, base_env: dict):
"""Launch the strategy subprocess."""
env = {**os.environ, **base_env, **self.env_overrides}
try:
self.process = subprocess.Popen(
[sys.executable, self.script_path],
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
self.state = StrategyState.STARTING
logger.info(f"[{self.name}] Started with PID {self.process.pid}")
except Exception as e:
logger.error(f"[{self.name}] Failed to start: {e}")
self.state = StrategyState.FAILED
def monitor(self):
"""Check process health and handle crashes."""
if self.process is None:
return
return_code = self.process.poll()
if return_code is not None:
# Process has exited
logger.warning(f"[{self.name}] Process exited with code {return_code}")
if self.restart_count < self.max_restart_attempts:
now = time.time()
# Enforce cooldown between restarts
if (self.last_restart_time is None or
now - self.last_restart_time >= self.restart_cooldown):
self.restart_count += 1
self.last_restart_time = now
logger.info(
f"[{self.name}] Restart attempt {self.restart_count}/"
f"{self.max_restart_attempts} in {self.restart_cooldown}s"
)
time.sleep(self.restart_cooldown)
return "restart"
else:
remaining = self.restart_cooldown - (now - self.last_restart_time)
logger.info(f"[{self.name}] Cooldown active: {remaining:.0f}s remaining")
else:
logger.critical(
f"[{self.name}] Max restart attempts reached. "
f"Strategy disabled until manual intervention."
)
self.state = StrategyState.FAILED
def stop(self):
"""Gracefully terminate the strategy process."""
if self.process:
logger.info(f"[{self.name}] Sending SIGTERM")
self.process.terminate()
try:
self.process.wait(timeout=10)
except subprocess.TimeoutExpired:
logger.warning(f"[{self.name}] SIGTERM timed out, sending SIGKILL")
self.process.kill()
self.state = StrategyState.STOPPED
class Supervisor:
"""
Manages multiple strategy processes with restart logic and health monitoring.
Run as the main orchestrator process on the server.
"""
def __init__(self, check_interval: float = 10.0):
self.strategies: dict[str, StrategyProcess] = {}
self.check_interval = check_interval
self.base_env: dict[str, str] = {}
self._running = False
self._monitor_thread: Optional[threading.Thread] = None
# Handle shutdown signals gracefully
signal.signal(signal.SIGTERM, self._handle_shutdown)
signal.signal(signal.SIGINT, self._handle_shutdown)
def register_strategy(self, name: str, script_path: str, env_overrides: dict = None):
"""Register a strategy to be supervised."""
self.strategies[name] = StrategyProcess(
name=name,
script_path=script_path,
env_overrides=env_overrides or {},
)
logger.info(f"Registered strategy: {name}")
def start_all(self):
"""Start all registered strategies."""
logger.info(f"Starting {len(self.strategies)} strategies...")
for strategy in self.strategies.values():
strategy.start(self.base_env)
self._running = True
self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
self._monitor_thread.start()
logger.info("Supervisor running")
def _monitor_loop(self):
"""Main monitoring loop: check health and restart failed strategies."""
while self._running:
for name, strategy in list(self.strategies.items()):
action = strategy.monitor()
if action == "restart":
strategy.start(self.base_env)
elif strategy.state == StrategyState.FAILED:
# In degraded mode: alert and wait for manual intervention
self._send_alert(f"Strategy {name} has failed and is disabled.")
time.sleep(self.check_interval)
def _handle_shutdown(self, signum, frame):
"""Graceful shutdown on SIGTERM or SIGINT."""
logger.info("Shutdown signal received, stopping all strategies...")
self._running = False
for strategy in self.strategies.values():
strategy.stop()
logger.info("Shutdown complete")
def _send_alert(self, message: str):
"""Send alert via configured notification channel."""
# Implementation: send to Telegram, PagerDuty, email, etc.
logger.critical(f"ALERT: {message}")
# ─────────────────────────────────────────────────────────────────────────────
# Example usage: start supervisor from main entry point
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
supervisor = Supervisor(check_interval=10.0)
# Register strategies with individual environment configs
supervisor.register_strategy(
name="mean_reversion",
script_path="/opt/strategies/mean_reversion.py",
env_overrides={
"STRATEGY_MODE": "live",
"MAX_POSITIONS": "5",
}
)
supervisor.register_strategy(
name="momentum",
script_path="/opt/strategies/momentum.py",
env_overrides={
"STRATEGY_MODE": "live",
"MAX_POSITIONS": "3",
}
)
supervisor.start_all()
# Keep main thread alive
while True:
time.sleep(60)
3.3 State Persistence: SQLite for Strategy State
Each strategy writes its state to a local SQLite database. This is not a performance bottleneck for individual strategies — SQLite handles thousands of writes per second easily on an SSD, and the write volume for a few strategies is well within that bound.
"""
Strategy state persistence layer using SQLite.
Provides atomic writes, queryable history, and easy backup.
"""
import sqlite3
import json
import time
from datetime import datetime, timezone
from contextlib import contextmanager
from pathlib import Path
from typing import Optional, Any
class StrategyStateDB:
"""
SQLite-backed state store for quantitative strategy persistence.
Schema:
- positions: current open positions
- orders: order history
- equity_curve: daily/account equity snapshots
- events: significant strategy events (starts, stops, errors)
"""
def __init__(self, db_path: str, strategy_name: str):
self.db_path = Path(db_path)
self.strategy_name = strategy_name
self.db_path.parent.mkdir(parents=True, exist_ok=True)
self._init_schema()
def _init_schema(self):
"""Initialize database schema if not exists."""
with self._conn() as conn:
conn.executescript("""
CREATE TABLE IF NOT EXISTS positions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
quantity REAL NOT NULL,
entry_price REAL,
current_price REAL,
unrealized_pnl REAL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id TEXT UNIQUE NOT NULL,
symbol TEXT NOT NULL,
side TEXT NOT NULL,
quantity REAL NOT NULL,
price REAL,
status TEXT NOT NULL,
created_at TEXT NOT NULL,
filled_at TEXT
);
CREATE TABLE IF NOT EXISTS equity_curve (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
equity REAL NOT NULL,
cash REAL NOT NULL,
positions_value REAL NOT NULL,
daily_return REAL
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
event_type TEXT NOT NULL,
message TEXT,
metadata TEXT
);
CREATE INDEX IF NOT EXISTS idx_orders_symbol ON orders(symbol);
CREATE INDEX IF NOT EXISTS idx_orders_created ON orders(created_at);
CREATE INDEX IF NOT EXISTS idx_equity_timestamp ON equity_curve(timestamp);
""")
@contextmanager
def _conn(self):
"""Thread-safe SQLite connection context manager."""
conn = sqlite3.connect(self.db_path, timeout=10.0, check_same_thread=False)
conn.execute("PRAGMA journal_mode=WAL") # Write-Ahead Logging for concurrent access
conn.execute("PRAGMA busy_timeout=5000")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def record_event(self, event_type: str, message: str, metadata: dict = None):
"""Log a significant strategy event."""
with self._conn() as conn:
conn.execute(
"""INSERT INTO events (timestamp, event_type, message, metadata)
VALUES (?, ?, ?, ?)""",
(
datetime.now(timezone.utc).isoformat(),
event_type,
message,
json.dumps(metadata) if metadata else None,
)
)
def update_position(self, symbol: str, quantity: float, entry_price: float,
current_price: float):
"""Update or insert a position."""
unrealized_pnl = (current_price - entry_price) * quantity
with self._conn() as conn:
conn.execute(
"""INSERT INTO positions
(symbol, quantity, entry_price, current_price, unrealized_pnl, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(symbol) DO UPDATE SET
quantity = excluded.quantity,
entry_price = excluded.entry_price,
current_price = excluded.current_price,
unrealized_pnl = excluded.unrealized_pnl,
updated_at = excluded.updated_at""",
(
symbol,
quantity,
entry_price,
current_price,
unrealized_pnl,
datetime.now(timezone.utc).isoformat(),
)
)
def record_equity(self, equity: float, cash: float, positions_value: float,
daily_return: Optional[float] = None):
"""Record an equity curve snapshot."""
with self._conn() as conn:
conn.execute(
"""INSERT INTO equity_curve
(timestamp, equity, cash, positions_value, daily_return)
VALUES (?, ?, ?, ?, ?)""",
(
datetime.now(timezone.utc).isoformat(),
equity,
cash,
positions_value,
daily_return,
)
)
def get_recent_equity(self, days: int = 30) -> list[dict]:
"""Retrieve recent equity curve data for analysis."""
with self._conn() as conn:
conn.row_factory = sqlite3.Row
cursor = conn.execute(
"""SELECT * FROM equity_curve
ORDER BY timestamp DESC LIMIT ?""",
(days,)
)
return [dict(row) for row in cursor.fetchall()]
4. Monitoring and Alerting: The Operations Layer
4.1 The Three Monitoring Pillars
A quantitative system needs three layers of monitoring. Each serves a distinct purpose:
| Pillar | Purpose | Tools |
|---|---|---|
| Uptime monitoring | Is the server reachable? Is the WebSocket connected? | Uptime Kuma, Pingdom |
| Process monitoring | Are strategy processes alive? Are they consuming expected resources? | Supervisor + custom scripts |
| Performance monitoring | Is the strategy performing as expected? Are latencies within bounds? | Prometheus + Grafana |
4.2 Lightweight Monitoring Stack
For a single-server deployment, full Kubernetes-grade observability is overkill. The following stack provides 90% of the insight at 10% of the complexity:
Uptime Kuma — Self-hosted uptime monitoring with alerting to Telegram or email. Monitor:
- Server HTTP endpoint (port 8000)
- WebSocket health endpoint in each strategy
- API response times
Prometheus node_exporter — System-level metrics (CPU, RAM, disk I/O, network). Runs as a sidecar on the same server.
Grafana — Visual dashboards for strategy equity curves, latency distributions, and system resource usage.
Custom health endpoint — Each strategy exposes a /health endpoint returning JSON:
from fastapi import FastAPI
import psutil
import logging
logger = logging.getLogger("HealthServer")
app = FastAPI()
@app.get("/health")
def health_check():
"""
Health endpoint for load balancer and monitoring probes.
Returns system and process health metrics.
"""
return {
"status": "healthy",
"strategy_name": os.environ.get("STRATEGY_NAME", "unknown"),
"uptime_seconds": time.time() - process_start_time,
"cpu_percent": psutil.cpu_percent(interval=0.1),
"memory_mb": psutil.Process().memory_info().rss / 1024 / 1024,
"data_freshness_seconds": (
datetime.now(timezone.utc) - last_data_timestamp
).total_seconds() if last_data_timestamp else None,
"active_positions": len(positions),
"equity": current_equity,
}
4.3 Alerting Rules and Thresholds
Set alerts at three severity levels:
| Severity | Condition | Action |
|---|---|---|
| Info | Data freshness > 30 seconds | Log warning, continue |
| Warning | Data freshness > 2 minutes | Telegram alert, strategy enters degraded mode |
| Critical | Process crash / data freshness > 10 minutes | Full alert, page developer |
| Emergency | Multiple consecutive failures, max retries exceeded | Disable strategy, alert with manual intervention required |
The critical discipline in alerting is avoiding alert fatigue. A strategy that generates 50 Telegram messages per day trains the developer to ignore alerts. A strategy that generates 1 alert per week gets immediate attention.
Calibrate thresholds accordingly:
- Do not alert on normal market microstructure events (normal spread widening, standard volatility).
- Alert on structural breaks: data feed dry, process crash, unexplained equity cliff.
5. Automated Operations: Deployment and Recovery
5.1 One-Command Deployment
Every infrastructure change should be deployable with a single command. If deployment requires more than one command, the process is too fragile for a system that runs unattended.
#!/bin/bash
# deploy.sh — Single-command deployment for the quantitative system
# Usage: ./deploy.sh [strategy_name | all]
set -euo pipefail
STRATEGY="${1:-all}"
SERVER_USER="quant"
SERVER_HOST="your-server.example.com"
REMOTE_DIR="/opt/strategies"
# Local build / validation
echo "[1/4] Running pre-deployment checks..."
python -m pytest tests/ --tb=short || { echo "Tests failed"; exit 1; }
echo "[2/4] Creating deployment archive..."
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
ARCHIVE="deploy_${TIMESTAMP}.tar.gz"
tar -czf "$ARCHIVE" \
--exclude='__pycache__' \
--exclude='*.pyc' \
--exclude='.git' \
--exclude='.env' \
strategies/ supervisor/ config/ requirements.txt
echo "[3/4] Uploading to server..."
scp "$ARCHIVE" ${SERVER_USER}@${SERVER_HOST}:/tmp/
echo "[4/4] Deploying on server..."
ssh ${SERVER_USER}@${SERVER_HOST} << 'REMOTE_SCRIPT'
set -e
ARCHIVE=$(ls -t /tmp/deploy_*.tar.gz | head -1)
BACKUP_DIR="/opt/strategies_backup_$(date +%Y%m%d_%H%M%S)"
# Backup current state
if [ -d /opt/strategies ]; then
cp -r /opt/strategies "$BACKUP_DIR"
echo "Backed up current state to $BACKUP_DIR"
fi
# Extract new deployment
tar -xzf "$ARCHIVE" -C /opt/strategies
cd /opt/strategies
# Install dependencies
pip install -r requirements.txt --quiet
# Restart supervisor
systemctl restart quant-supervisor || {
echo "Supervisor restart failed — manual intervention required"
exit 1
}
echo "Deployment complete"
REMOTE_SCRIPT
echo "Deployment successful at $(date)"
rm "$ARCHIVE"
5.2 Automatic Recovery Patterns
The system must recover from common failure modes without human intervention:
| Failure mode | Automatic recovery |
|---|---|
| Process crash | Supervisor restarts with exponential backoff |
| WebSocket disconnection | Client reconnects automatically with backoff |
| Out-of-memory | Supervisor kills and restarts with reduced position limits |
| Disk full | Alert sent; strategy halts trading until resolved |
| API rate limit | Exponential backoff with jitter; no data loss |
| Market data gap | Strategy enters degraded mode; no new orders until data resumes |
The goal is not zero failures. The goal is zero manual interventions. A system that recovers automatically is more reliable than a system that never crashes but requires human attention every time anything goes wrong.
6. Cost Optimization: The $40-per-Month Discipline
6.1 Where the Money Goes — and Where It Does Not
Most cloud waste comes from three sources: overprovisioned compute, idle resources, and premium support tiers you do not need.
For a single-server quantitative system:
| Resource | Right-size | Oversized |
|---|---|---|
| CPU | 4 dedicated vCPU | 8 vCPU (diminishing returns for strategy Python) |
| RAM | 8 GB | 16 GB (no benefit unless running in-memory databases) |
| Storage | 100 GB SSD | 500 GB (cost of storage rarely matters at this scale) |
| Network | Standard | Premium egress (you pay for data leaving, not entering) |
| Support | Community | Paid support (you are debugging your own code) |
6.2 Data Cost Architecture
Market data costs vary dramatically by provider and data type. Design your data consumption to match your actual trading frequency:
| Data type | Typical cost | Use case |
|---|---|---|
| Real-time WebSocket | Free to $50/month | Live strategy execution |
| Historical OHLCV | $0–500/month | Backtesting and strategy development |
| Level 2 depth | $100–1000/month | Advanced order book strategies |
| Full tick data | $1000+/month | HFT research only |
A personal developer running one to three strategies should start with real-time WebSocket data and historical OHLCV for backtesting. Do not pay for depth data until you have a strategy that demonstrably requires it.
TickDB's pricing model for US equity OHLCV data is structured for exactly this use case: free or low-cost access for development and backtesting, with paid tiers for extended historical data. This aligns the cost structure with the actual development lifecycle.
6.3 Monthly Cost Estimate
| Component | Monthly cost |
|---|---|
| Cloud server (4 vCPU, 8 GB, 100 GB SSD) | $40 |
| Domain (subdomain of free DDNS) | $0 |
| SSL certificate (Let's Encrypt) | $0 |
| Monitoring (self-hosted Uptime Kuma + Grafana) | $0 |
| Market data (TickDB WebSocket — development tier) | $0–$25 |
| Notification (Telegram) | $0 |
| Total | $40–65/month |
A system running at this cost level is not a toy. It is a production system with real resilience, real monitoring, and real market data. The constraint is not technical — it is psychological. You must accept that the system will occasionally need human attention, and you must build the monitoring and alerting to minimize how often that happens.
7. Deployment Guide by Experience Level
The right starting point depends on where you are in your quant development journey.
| Stage | Starting point | Growth path |
|---|---|---|
| New to Python / trading | Single strategy, REST polling every 30 seconds, single .py file |
Add WebSocket streaming; split into modules |
| Building first strategy | WebSocket data client + SQLite state + supervisor | Add monitoring stack; implement backtesting module |
| Running live strategies | Full production stack as described in this article | Add more strategies; implement portfolio-level risk management |
| Institutional aspirations | Migrate to dedicated VPS or co-located server; add Kubernetes for orchestration | This article is your foundation, not your ceiling |
8. Closing
The system described in this article runs on a $40-per-month cloud server. It has automatic reconnection, process isolation, state persistence, health monitoring, and alerting. It survives the kinds of failures that destroy manual trading setups — connectivity drops, API rate limits, process crashes, overnight gaps.
The architecture is not glamorous. It does not use Kubernetes. It does not have a microservices mesh. It has one server, a supervisor, a few strategy processes, and well-written monitoring.
That is the point.
The best infrastructure is the infrastructure you can understand, debug, and improve at 2 AM. For an individual quant developer, that is a single well-engineered server, not a distributed system built for teams of 20.
Start simple. Measure everything. Add complexity only when you have a specific, measurable reason to do so.
The code in this article is designed to be read, understood, and modified. Copy it. Break it. Fix it. Make it yours.
That is how a single developer builds a quantitative system that actually survives contact with the markets.
Next Steps
If you are just starting out: Visit TickDB API documentation and claim a free API key. The code in this article is designed to work with any market data provider, but starting with a known-good data source reduces debugging surface area significantly.
If you are ready to deploy this system: Use the deployment script in Section 5.1 as your starting template. Replace the placeholder values with your actual server credentials and strategy paths.
If you need extended historical data for backtesting: TickDB's Professional plan provides 10+ years of cleaned US equity OHLCV data, suitable for multi-year backtests across different market regimes. Evaluate whether your strategy's edge survives across bull markets, bear markets, and low-volatility crushes before committing capital.
If you use AI coding assistants: Search for and install the tickdb-market-data SKILL in your AI tool's marketplace. It provides pre-built code templates, endpoint documentation, and best-practice patterns for integrating real-time market data into automated trading systems.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Quantitative trading strategies can incur significant losses. Always validate with paper trading before committing capital.