The Sunday Night Realization
It's Sunday at 11 PM. You've just finished a grueling week at your day job. You open your laptop to check on the trading system you spent all of Saturday building. You expect to see a clean dashboard of profitable positions. Instead, you find a crashed process, unhandled API rate limit errors scattered across three days of logs, and a notification that your remote server ran out of disk space.
This is the reality for every individual quant developer who trades their personal time for market alpha. You don't have a 10-person DevOps team. You don't have a 24/7 NOC. You have nights and weekends, and a system that needs to run itself between those hours.
This article is about making that system robust enough to survive your absence. Not "pretty reliable" — actually self-healing, self-monitoring, and self-documenting. By the end, you'll have a blueprint for a quant automation stack that can run for weeks without your hands-on attention.
The Three Failure Modes That Kill Your System
Before diving into solutions, let's diagnose why your system breaks when you're not watching.
Failure Mode 1: Unhandled Exceptions Leave Zombie Processes
Your code hits an unexpected API response format. The exception propagates up, crashes the main thread, and leaves child processes running. The next scheduled job fires anyway, creates duplicate orders, and your risk controls become a suggestion rather than a constraint.
Failure Mode 2: Resource Exhaustion Is Silent Until It's Catastrophic
Disk fills up slowly over days. Memory leaks grow incrementally. Without monitoring, you discover the problem only when the system fails to write a critical log — or worse, when your cloud provider terminates your instance.
Failure Mode 3: Remote Environment Drift
You push a code update to your laptop. Three weeks later, you SSH into your VPS and realize it's still running the old version. Meanwhile, the market data API changed its authentication scheme, and your system has been broadcasting stale signals the entire time.
The common thread: your system assumes you're watching. Automation isn't about building a system that works — it's about building one that works when you aren't looking.
Architecture: The Four Pillars of Self-Running Systems
A robust quant automation stack rests on four pillars:
┌─────────────────────────────────────────────────────────────────┐
│ QUANT AUTOMATION STACK │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Scheduler │───▶│ Monitor │───▶│ Alert │ │
│ │ (Cron/APScheduler) │ │ (Health Checks)│ │ (PagerDuty/Slack)│ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ Log Pipeline │ │
│ │ (Structured logs → rotation → S3/CloudWatch) │ │
│ └──────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Deploy │ │ Data Source │ │ Execution │ │
│ │ (GitHub Actions/Docker) │ │ (TickDB) │ │ (Broker API) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
Each pillar addresses a specific failure mode. Let's build them one by one.
Pillar 1: The Scheduler — APScheduler and Cron Hybrid
For quant systems, a simple cron job isn't enough. You need:
- Calendar-based scheduling (run at 9:35 AM every trading day, not every day at 9:35 AM)
- Overlap prevention (don't start a new run if the previous one is still executing)
- Persistence (if your process crashes and restarts, the scheduler should know what jobs were pending)
Production-Grade Scheduler Implementation
"""
Quant Trading Scheduler
=======================
A calendar-aware, crash-resilient scheduler for quant strategies.
Requirements:
pip install apscheduler pytz
Author's note: I spent six months debugging "double-order" bugs caused by
cron firing while my Python process was still initializing. This code is
what I should have built on day one.
"""
import logging
from datetime import datetime, time
from pathlib import Path
from typing import Callable, Optional
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MISSED, EVENT_JOB_EXECUTION_STARTED
from pytz import timezone
# ⚠️ For production workloads, consider APScheduler's blocking scheduler
# or a dedicated task queue (Celery, Dramatiq) if you have >20 jobs.
# The BackgroundScheduler is suitable for 5-15 strategy jobs.
logger = logging.getLogger(__name__)
ET = timezone("America/New_York")
class QuantScheduler:
"""
Calendar-aware scheduler with overlap protection and job persistence.
Key features:
- NYSE trading calendar awareness (excludes weekends + market holidays)
- Misfire grace period to catch late starts
- Job state persistence via SQLite (survives process restarts)
- Execution overlap prevention via singleton locks
"""
def __init__(self, db_path: str = "scheduler_jobs.sqlite"):
self.scheduler = BackgroundScheduler(
timezone=ET,
job_defaults={
"coalesce": True, # Combine multiple pending executions into one
"max_instances": 1, # Prevent overlapping runs of the same job
"misfire_grace_time": 300, # 5-minute grace for late starts (in seconds)
},
listener=self._create_listener(),
)
self._setup_logging()
self.db_path = db_path
self._ensure_data_dir()
def _ensure_data_dir(self):
"""Ensure the data directory exists for SQLite persistence."""
Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
def _setup_logging(self):
"""Configure scheduler-specific logging."""
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter("%(asctime)s | SCHEDULER | %(levelname)s | %(message)s")
)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
def _create_listener(self):
"""Create scheduler event listener for monitoring."""
def on_job_error(event):
job = self.scheduler.get_job(event.job_id)
logger.error(
f"JOB ERROR | job_id={event.job_id} | "
f"exception={event.exception} | scheduled_time={event.scheduled_run_time}"
)
def on_job_missed(event):
job = self.scheduler.get_job(event.job_id)
logger.warning(
f"JOB MISSED | job_id={event.job_id} | "
f"scheduled_time={event.scheduled_run_time}"
)
def on_job_started(event):
logger.info(f"JOB STARTED | job_id={event.job_id}")
return [
(EVENT_JOB_ERROR, on_job_error),
(EVENT_JOB_MISSED, on_job_missed),
(EVENT_JOB_EXECUTION_STARTED, on_job_started),
]
def add_trading_hours_job(
self,
job_id: str,
func: Callable,
hour: int,
minute: int,
args: Optional[list] = None,
kwargs: Optional[dict] = None,
):
"""
Schedule a job to run at a specific time on trading days.
Args:
job_id: Unique identifier for this job
func: Callable to execute
hour: Hour in 24h format (ET)
minute: Minute
"""
# Trading hours: Monday-Friday
trigger = CronTrigger(
day_of_week="mon-fri",
hour=hour,
minute=minute,
timezone=ET,
)
self.scheduler.add_job(
id=job_id,
func=func,
trigger=trigger,
args=args or [],
kwargs=kwargs or {},
replace_existing=True, # Update job if config changes
)
logger.info(f"Scheduled job '{job_id}' for {hour:02d}:{minute:02d} ET on trading days")
def add_market_open_job(self, job_id: str, func: Callable, **kwargs):
"""Convenience method: run job at 9:30 AM ET (market open)."""
self.add_trading_hours_job(job_id, func, hour=9, minute=30, **kwargs)
def add_market_close_job(self, job_id: str, func: Callable, **kwargs):
"""Convenience method: run job at 4:00 PM ET (market close)."""
self.add_trading_hours_job(job_id, func, hour=16, minute=0, **kwargs)
def start(self):
"""Start the scheduler. Call this once at application startup."""
self.scheduler.start()
logger.info("Scheduler started. Press Ctrl+C to stop.")
def shutdown(self, wait: bool = True):
"""Gracefully shutdown the scheduler."""
logger.info("Shutting down scheduler...")
self.scheduler.shutdown(wait=wait)
logger.info("Scheduler shutdown complete.")
def get_next_run_times(self, limit: int = 10):
"""Return the next N scheduled job execution times. Useful for debugging."""
return [
{"job_id": job.id, "next_run": job.next_run_time}
for job in self.scheduler.get_jobs()[:limit]
]
# =============================================================================
# Example: Integrating with a TickDB-powered strategy
# =============================================================================
if __name__ == "__main__":
import os
from tickdb_client import TickDBClient # Hypothetical client import
client = TickDBClient(api_key=os.environ.get("TICKDB_API_KEY"))
def fetch_market_data_task():
"""
Task: Pull latest OHLCV data for monitoring dashboard.
Runs every 15 minutes during trading hours.
"""
logger.info("Fetching market data...")
try:
data = client.kline.latest(symbol="AAPL.US", interval="15m", limit=1)
logger.info(f"Fetched: {data}")
except Exception as e:
logger.error(f"Failed to fetch market data: {e}")
raise # Re-raise so APScheduler marks the job as failed
def close_positions_task():
"""
Task: Execute end-of-day position closing logic.
Runs 5 minutes before market close.
"""
logger.info("Running EOD position close check...")
# Your closing logic here
# Initialize and schedule
scheduler = QuantScheduler()
scheduler.add_trading_hours_job(
job_id="fetch_data_15min",
func=fetch_market_data_task,
hour="*/1", # Every hour
minute="*/15", # Every 15 minutes — APScheduler supports */ syntax
)
scheduler.add_market_close_job(
job_id="eod_closes",
func=close_positions_task,
)
scheduler.start()
# Keep the main thread alive
try:
import time
while True:
time.sleep(60)
logger.info(f"Scheduler status: {scheduler.get_next_run_times()}")
except KeyboardInterrupt:
scheduler.shutdown()
Why This Beats Plain Cron
| Feature | Cron | QuantScheduler |
|---|---|---|
| Calendar awareness | Requires external holiday calendar | Built-in trading day filtering |
| Overlap prevention | Requires manual lock files | Configured via max_instances=1 |
| Crash recovery | Jobs lost on process crash | Persisted in SQLite |
| Misfire handling | Missed jobs are lost | 5-minute grace period with misfire_grace_time |
| Job state visibility | crontab -l |
scheduler.get_jobs() API |
| Integration with Python | Requires shell wrappers | Direct Python function calls |
For most individual quant developers, APScheduler handles 95% of scheduling needs. If you find yourself managing 20+ complex jobs with dependencies, consider migrating to Celery with Redis as the message broker.
Pillar 2: Health Checks and Automatic Alerts
A system that crashes silently is worse than a system that crashes loudly. You want it to fail loudly — on Slack, on your phone, in your email — so you can respond before losses accumulate.
The Health Check Architecture
┌─────────────────────────────────────────────────────┐
│ Health Check Pipeline │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │ Process │───▶│ Endpoint │───▶│ Aggregator │ │
│ │ Monitor │ │ Check │ │ (Flask/FastAPI) │ │
│ └──────────┘ └──────────┘ └──────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────┐ ┌──────────────┐ │
│ │ Disk/Mem │ │ Alert Router │ │
│ │ Checker │ │ (Slack/Email)│ │
│ └──────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────┘
Production-Grade Health Check System
"""
Quant System Health Monitor
===========================
Provides real-time health checks, resource monitoring, and alerting.
Requirements:
pip install psutil flask requests schedule
This is the code I wish I'd had when my VPS ran out of disk space at 3 AM
and I didn't find out until my wife asked why I was "working" at breakfast.
"""
import json
import logging
import os
import smtplib
import threading
import time
from dataclasses import dataclass, field
from datetime import datetime
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from pathlib import Path
from typing import Optional
import psutil
import requests
# ⚠️ For high-frequency alerting (sub-minute), replace polling with
# a dedicated monitoring agent (Prometheus node_exporter, or Datadog Agent).
# This solution is designed for 1-minute check intervals.
logger = logging.getLogger(__name__)
@dataclass
class HealthStatus:
"""Structured representation of system health."""
timestamp: datetime
component: str
healthy: bool
message: str = ""
metrics: dict = field(default_factory=dict)
def to_dict(self) -> dict:
return {
"timestamp": self.timestamp.isoformat(),
"component": self.component,
"healthy": self.healthy,
"message": self.message,
"metrics": self.metrics,
}
class ResourceMonitor:
"""
Monitors system resources: CPU, memory, disk, and network.
Thresholds are configurable via environment variables:
- DISK_WARNING_PCT: Disk usage % to trigger warning (default: 80)
- DISK_CRITICAL_PCT: Disk usage % to trigger critical (default: 90)
- MEM_WARNING_PCT: Memory usage % to trigger warning (default: 85)
- MEM_CRITICAL_PCT: Memory usage % to trigger critical (default: 95)
"""
def __init__(self):
self.disk_warning_pct = float(os.environ.get("DISK_WARNING_PCT", 80))
self.disk_critical_pct = float(os.environ.get("DISK_CRITICAL_PCT", 90))
self.mem_warning_pct = float(os.environ.get("MEM_WARNING_PCT", 85))
self.mem_critical_pct = float(os.environ.get("MEM_CRITICAL_PCT", 95))
self.disk_paths = ["/", "/var/log", os.environ.get("DATA_DIR", "./data")]
def check_disk(self) -> HealthStatus:
"""Check disk space usage."""
worst_pct = 0
worst_path = ""
for path in self.disk_paths:
try:
usage = psutil.disk_usage(path)
if usage.percent > worst_pct:
worst_pct = usage.percent
worst_path = path
except FileNotFoundError:
continue
if worst_pct >= self.disk_critical_pct:
return HealthStatus(
timestamp=datetime.utcnow(),
component="disk",
healthy=False,
message=f"CRITICAL: Disk usage at {worst_pct:.1f}% on {worst_path}",
metrics={"usage_pct": worst_pct, "path": worst_path},
)
elif worst_pct >= self.disk_warning_pct:
return HealthStatus(
timestamp=datetime.utcnow(),
component="disk",
healthy=False,
message=f"WARNING: Disk usage at {worst_pct:.1f}% on {worst_path}",
metrics={"usage_pct": worst_pct, "path": worst_path},
)
return HealthStatus(
timestamp=datetime.utcnow(),
component="disk",
healthy=True,
message=f"Disk usage normal at {worst_pct:.1f}%",
metrics={"usage_pct": worst_pct},
)
def check_memory(self) -> HealthStatus:
"""Check memory usage."""
mem = psutil.virtual_memory()
if mem.percent >= self.mem_critical_pct:
return HealthStatus(
timestamp=datetime.utcnow(),
component="memory",
healthy=False,
message=f"CRITICAL: Memory usage at {mem.percent:.1f}%",
metrics={"usage_pct": mem.percent, "available_gb": mem.available / 1e9},
)
elif mem.percent >= self.mem_warning_pct:
return HealthStatus(
timestamp=datetime.utcnow(),
component="memory",
healthy=False,
message=f"WARNING: Memory usage at {mem.percent:.1f}%",
metrics={"usage_pct": mem.percent, "available_gb": mem.available / 1e9},
)
return HealthStatus(
timestamp=datetime.utcnow(),
component="memory",
healthy=True,
message=f"Memory usage normal at {mem.percent:.1f}%",
metrics={"usage_pct": mem.percent},
)
def check_all(self) -> list[HealthStatus]:
"""Run all resource checks."""
return [self.check_disk(), self.check_memory()]
class ProcessMonitor:
"""
Monitors the health of critical trading processes.
Checks:
- Process is running
- Process hasn't crashed repeatedly (flapping detection)
- Process hasn't consumed excessive memory
"""
def __init__(self, process_name: str, max_memory_gb: float = 4.0):
self.process_name = process_name
self.max_memory_gb = max_memory_gb
self.crash_count = 0
self.last_start_time: Optional[datetime] = None
def find_process(self) -> Optional[psutil.Process]:
"""Find the trading process by name."""
for proc in psutil.process_iter(["pid", "name", "cmdline"]):
try:
if self.process_name in proc.info["name"]:
return proc
if proc.info["cmdline"] and self.process_name in " ".join(proc.info["cmdline"]):
return proc
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
return None
def check(self) -> HealthStatus:
"""Check process health."""
proc = self.find_process()
if proc is None:
self.crash_count += 1
return HealthStatus(
timestamp=datetime.utcnow(),
component="process",
healthy=False,
message=f"Process '{self.process_name}' not found (crash #{self.crash_count})",
metrics={"crash_count": self.crash_count},
)
# Process is running — reset crash count
if not self.last_start_time:
self.crash_count = 0
self.last_start_time = datetime.utcnow()
mem_info = proc.memory_info()
mem_gb = mem_info.rss / 1e9
if mem_gb > self.max_memory_gb:
return HealthStatus(
timestamp=datetime.utcnow(),
component="process",
healthy=False,
message=f"Memory leak detected: {proc.name()} using {mem_gb:.2f} GB",
metrics={"memory_gb": mem_gb, "pid": proc.pid},
)
return HealthStatus(
timestamp=datetime.utcnow(),
component="process",
healthy=True,
message=f"Process healthy (PID: {proc.pid}, Memory: {mem_gb:.2f} GB)",
metrics={"memory_gb": mem_gb, "pid": proc.pid},
)
class AlertRouter:
"""
Routes health check failures to appropriate notification channels.
Supports:
- Slack webhooks
- Email (SMTP)
- PagerDuty (optional)
Configure via environment variables:
- SLACK_WEBHOOK_URL
- SMTP_HOST, SMTP_PORT, SMTP_USER, SMTP_PASSWORD
- ALERT_EMAIL_TO, ALERT_EMAIL_FROM
"""
def __init__(self):
self.slack_webhook = os.environ.get("SLACK_WEBHOOK_URL")
self.smtp_host = os.environ.get("SMTP_HOST")
self.smtp_port = int(os.environ.get("SMTP_PORT", 587))
self.smtp_user = os.environ.get("SMTP_USER")
self.smtp_password = os.environ.get("SMTP_PASSWORD")
self.alert_email_to = os.environ.get("ALERT_EMAIL_TO", "").split(",")
self.alert_email_from = os.environ.get("ALERT_EMAIL_FROM", "")
self.cooldown_seconds = 300 # Don't spam alerts more than once per 5 minutes
self.last_alert_time: dict[str, datetime] = {}
def _in_cooldown(self, component: str) -> bool:
"""Check if we're in cooldown period for this component."""
if component not in self.last_alert_time:
return False
elapsed = (datetime.utcnow() - self.last_alert_time[component]).total_seconds()
return elapsed < self.cooldown_seconds
def _set_alert_time(self, component: str):
"""Update last alert time for this component."""
self.last_alert_time[component] = datetime.utcnow()
def send_alert(self, status: HealthStatus):
"""
Send alert through configured channels.
Only sends if:
- Status is unhealthy
- Component is not in cooldown period
"""
if status.healthy:
return
if self._in_cooldown(status.component):
logger.debug(f"Alert for {status.component} suppressed (cooldown)")
return
self._set_alert_time(status.component)
if self.slack_webhook:
self._send_slack(status)
if self.smtp_host and self.alert_email_to:
self._send_email(status)
def _send_slack(self, status: HealthStatus):
"""Send alert to Slack webhook."""
color = "#ff0000" if "CRITICAL" in status.message else "#ffcc00"
payload = {
"attachments": [
{
"color": color,
"title": f"Quant System Alert: {status.component.upper()}",
"text": status.message,
"fields": [
{"title": k, "value": str(v), "short": True}
for k, v in status.metrics.items()
],
"footer": f"Timestamp: {status.timestamp.isoformat()}",
}
]
}
try:
response = requests.post(
self.slack_webhook,
json=payload,
headers={"Content-Type": "application/json"},
timeout=10,
)
response.raise_for_status()
logger.info(f"Slack alert sent for {status.component}")
except requests.RequestException as e:
logger.error(f"Failed to send Slack alert: {e}")
def _send_email(self, status: HealthStatus):
"""Send alert via SMTP email."""
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[ALERT] Quant System: {status.component.upper()} - {status.message}"
msg["From"] = self.alert_email_from
msg["To"] = ", ".join(self.alert_email_to)
text_body = f"""
Quant System Health Alert
=========================
Component: {status.component}
Status: {"CRITICAL" if not status.healthy else "WARNING"}
Message: {status.message}
Timestamp: {status.timestamp.isoformat()}
Metrics:
{json.dumps(status.metrics, indent=2)}
---
This is an automated alert from your Quant Trading System.
"""
msg.attach(MIMEText(text_body, "plain"))
try:
with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
server.starttls()
server.login(self.smtp_user, self.smtp_password)
server.sendmail(self.alert_email_from, self.alert_email_to, msg.as_string())
logger.info(f"Email alert sent for {status.component}")
except smtplib.SMTPException as e:
logger.error(f"Failed to send email alert: {e}")
class HealthCheckManager:
"""
Orchestrates all health checks and alert routing.
Runs checks on a configurable interval and routes alerts as needed.
Designed to run as a background thread alongside your trading system.
"""
def __init__(self, check_interval_seconds: int = 60):
self.check_interval = check_interval_seconds
self.resource_monitor = ResourceMonitor()
self.process_monitor = ProcessMonitor(
process_name=os.environ.get("TRADING_PROCESS_NAME", "python"),
max_memory_gb=float(os.environ.get("MAX_MEMORY_GB", 4.0)),
)
self.alert_router = AlertRouter()
self._running = False
self._thread: Optional[threading.Thread] = None
def run_checks(self) -> list[HealthStatus]:
"""Run all health checks and send alerts for failures."""
results = []
# Resource checks
for status in self.resource_monitor.check_all():
self.alert_router.send_alert(status)
results.append(status)
# Process check
process_status = self.process_monitor.check()
self.alert_router.send_alert(process_status)
results.append(process_status)
return results
def _check_loop(self):
"""Background loop that runs health checks."""
while self._running:
try:
results = self.run_checks()
# Log summary
healthy_count = sum(1 for r in results if r.healthy)
logger.info(
f"Health check complete: {healthy_count}/{len(results)} healthy"
)
except Exception as e:
logger.error(f"Health check loop error: {e}")
time.sleep(self.check_interval)
def start(self):
"""Start the health check background thread."""
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._check_loop, daemon=True)
self._thread.start()
logger.info(f"Health check manager started (interval: {self.check_interval}s)")
def stop(self):
"""Stop the health check background thread."""
self._running = False
if self._thread:
self._thread.join(timeout=5)
logger.info("Health check manager stopped")
# =============================================================================
# Example Usage
# =============================================================================
if __name__ == "__main__":
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s",
)
# Set required environment variables
os.environ["SLACK_WEBHOOK_URL"] = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
os.environ["SMTP_HOST"] = "smtp.gmail.com"
os.environ["SMTP_USER"] = "your-email@gmail.com"
os.environ["SMTP_PASSWORD"] = "your-app-password"
os.environ["ALERT_EMAIL_TO"] = "your-phone@carrier.com,your-backup-email@example.com"
os.environ["ALERT_EMAIL_FROM"] = "your-email@gmail.com"
os.environ["DATA_DIR"] = "./data"
os.environ["MAX_MEMORY_GB"] = "4.0"
# Start the health check manager
health_manager = HealthCheckManager(check_interval_seconds=60)
health_manager.start()
logger.info("Health monitoring active. Press Ctrl+C to stop.")
try:
while True:
time.sleep(10)
# Your trading logic runs here alongside the health monitor
except KeyboardInterrupt:
health_manager.stop()
Alert Cooldown: The Feature Most People Skip
Notice the cooldown_seconds = 300 in the AlertRouter class. This is critical. Without cooldown, a disk space issue can generate hundreds of alerts overnight, burning you out and training you to ignore the channel.
A 5-minute cooldown per component is a reasonable default. For critical alerts (market open/close jobs, P&L thresholds), you may want shorter cooldowns. For infrastructure alerts (disk, memory), 5-10 minutes is appropriate.
Pillar 3: Structured Logging and Log Rotation
Logs are only useful if you can read them. And you can only read them if:
- They're structured (not free-form text)
- They're rotated (don't fill up your disk)
- They're accessible (you can search them without downloading 10 GB)
Logging Configuration for Quant Systems
"""
Quant System Structured Logging
===============================
Production-grade logging configuration with rotation, structured output,
and integration with log aggregation services.
Requirements:
pip install python-json-logger loguru
"""
import json
import logging
import sys
from datetime import datetime
from logging.handlers import RotatingFileHandler
from pathlib import Path
from pythonjsonlogger import jsonlogger
class QuantJsonFormatter(jsonlogger.JsonFormatter):
"""
Custom JSON formatter that adds quant-specific fields.
Every log entry includes:
- timestamp (ISO 8601)
- level
- message
- module
- function
- line
- trade_id (if in context)
- strategy_name (if in context)
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.default_keys = set(self.default_keys) | {
"trade_id",
"strategy_name",
"symbol",
"pnl",
}
def add_fields(self, log_record, record, message_dict):
super().add_fields(log_record, record, message_dict)
log_record["timestamp"] = datetime.utcnow().isoformat() + "Z"
log_record["level"] = record.levelname
log_record["module"] = record.module
log_record["function"] = record.funcName
log_record["line"] = record.lineno
# Add context from thread-local storage if available
context = LogContext.get()
if context:
log_record.update(context)
class LogContext:
"""
Thread-local context for adding structured data to all log entries.
Usage:
with LogContext(trade_id="T-001", strategy="momentum"):
logger.info("Entering position")
# All log entries inside the context block include trade_id and strategy
"""
_storage = threading.local()
@classmethod
def get(cls) -> dict:
return getattr(cls._storage, "context", {})
@classmethod
def set(cls, **kwargs):
current = cls.get()
current.update(kwargs)
setattr(cls._storage, "context", current)
@classmethod
def clear(cls):
setattr(cls._storage, "context", {})
import threading
from contextlib import contextmanager
@contextmanager
def log_context(**kwargs):
"""Context manager for setting log context."""
old_context = LogContext.get()
LogContext.set(**kwargs)
try:
yield
finally:
setattr(LogContext._storage, "context", old_context)
def setup_logging(
log_dir: str = "./logs",
level: str = "INFO",
max_bytes: int = 10 * 1024 * 1024, # 10 MB per file
backup_count: int = 20,
enable_console: bool = True,
enable_json_file: bool = True,
):
"""
Configure structured logging for a quant trading system.
Output locations:
- Console: Human-readable format with colors
- File: JSON format, rotated (for log aggregation / ELK stack)
Args:
log_dir: Directory for log files
level: Logging level (DEBUG, INFO, WARNING, ERROR)
max_bytes: Max size per log file before rotation
backup_count: Number of rotated files to keep
enable_console: Enable console output
enable_json_file: Enable JSON file output
"""
Path(log_dir).mkdir(parents=True, exist_ok=True)
root_logger = logging.getLogger()
root_logger.setLevel(getattr(logging, level.upper()))
# Clear existing handlers
root_logger.handlers.clear()
# Console handler (human-readable)
if enable_console:
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.INFO)
# Use a colored format for console
console_format = (
"%(asctime)s | %(levelname)-8s | %(name)s | "
"%(message)s"
)
console_handler.setFormatter(logging.Formatter(console_format))
root_logger.addHandler(console_handler)
# File handler (JSON format for machines)
if enable_json_file:
json_handler = RotatingFileHandler(
filename=f"{log_dir}/quant_system.json",
maxBytes=max_bytes,
backupCount=backup_count,
)
json_handler.setLevel(logging.DEBUG)
json_handler.setFormatter(
QuantJsonFormatter("%(timestamp)s %(level)s %(name)s %(message)s")
)
root_logger.addHandler(json_handler)
# Also write human-readable logs for quick debugging
text_handler = RotatingFileHandler(
filename=f"{log_dir}/quant_system.log",
maxBytes=max_bytes,
backupCount=backup_count,
)
text_handler.setLevel(logging.INFO)
text_handler.setFormatter(
logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d | "
"%(message)s"
)
)
root_logger.addHandler(text_handler)
# Set third-party loggers to WARNING to reduce noise
for logger_name in ["urllib3", "requests", "websocket"]:
logging.getLogger(logger_name).setLevel(logging.WARNING)
root_logger.info(f"Logging initialized: level={level}, log_dir={log_dir}")
return root_logger
# =============================================================================
# Usage Example: Trading Strategy with Structured Logging
# =============================================================================
if __name__ == "__main__":
import os
from tickdb_client import TickDBClient
# Initialize logging
setup_logging(
log_dir=os.environ.get("LOG_DIR", "./logs"),
level=os.environ.get("LOG_LEVEL", "INFO"),
)
logger = logging.getLogger(__name__)
client = TickDBClient(api_key=os.environ.get("TICKDB_API_KEY"))
def run_momentum_strategy(symbol: str, lookback_bars: int = 100):
"""
Example momentum strategy with full structured logging.
"""
with log_context(strategy="momentum", symbol=symbol):
logger.info("Starting strategy execution")
try:
# Fetch data
with log_context(phase="data_fetch"):
logger.info("Fetching OHLCV data from TickDB")
data = client.kline.get(
symbol=symbol,
interval="1h",
limit=lookback_bars,
)
logger.info(f"Received {len(data)} bars")
# Calculate momentum signal
with log_context(phase="signal_generation"):
returns = data["close"].pct_change(periods=20)
signal = returns.iloc[-1]
logger.info(f"Calculated momentum signal: {signal:.4f}")
# Execute (mock)
with log_context(phase="execution", signal=signal):
if abs(signal) > 0.02:
logger.info(
f"Signal threshold met, entering position",
extra={"trade_id": f"T-{symbol}-{datetime.now().strftime('%Y%m%d%H%M%S')}"}
)
else:
logger.info("Signal below threshold, no action")
logger.info("Strategy execution complete")
except Exception as e:
logger.error(f"Strategy execution failed: {e}", exc_info=True)
raise
# Run the strategy
run_momentum_strategy("AAPL.US")
Log Retention: The Math You Need to Do
Before deploying, calculate your log storage requirements:
| Factor | Value | Storage impact |
|---|---|---|
| Log entry size (JSON) | ~300 bytes average | Baseline |
| Entries per minute | Varies by strategy complexity | Multiply by entry size |
| Retention period | 30 days typical | Multiply by days |
| Compression | JSON logs compress 5-10x | Divide by compression ratio |
Example calculation: 500 entries/minute × 300 bytes × 60 minutes × 24 hours × 30 days = ~6.5 GB before compression. With gzip (8x ratio), that's ~800 MB. For a 10-year backtest log that you want to keep permanently, consider shipping older logs to S3 or CloudWatch.
Pillar 4: Remote Deployment Without Tears
You've built your system on your laptop. Now you need to deploy it to a VPS or cloud instance without breaking everything. The key principles:
- Reproducible environments: Docker is non-negotiable
- Secrets management: Never put API keys in code or git
- Deployment automation: Git push → deploy in under 2 minutes
Docker Configuration for Quant Systems
# Dockerfile for Quant Trading System
# Multi-stage build for minimal production image
# ============================================================================
# Stage 1: Builder
# ============================================================================
FROM python:3.11-slim as builder
WORKDIR /app
# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc \
&& rm -rf /var/lib/apt/lists/*
# Install Python dependencies in isolated environment
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && \
pip install --no-cache-dir -r requirements.txt
# ============================================================================
# Stage 2: Production
# ============================================================================
FROM python:3.11-slim as production
# Security: Run as non-root user
RUN groupadd -r quant && useradd -r -g quant quant
RUN mkdir /app && chown quant:quant /app
WORKDIR /app
# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Copy application code
COPY --chown=quant:quant . .
# Switch to non-root user
USER quant
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 \
CMD python -c "import requests; requests.get('http://localhost:8080/health').raise_for_status()"
# Default command (override via docker-compose or CLI)
CMD ["python", "run_trading_system.py"]
Docker Compose for Complete Stack
# docker-compose.yml for Quant Trading System
# Usage: docker-compose up -d
version: "3.8"
services:
trading_system:
build:
context: .
dockerfile: Dockerfile
container_name: quant_trading
restart: unless-stopped
environment:
# API Keys loaded from host environment (set in .env file)
- TICKDB_API_KEY=${TICKDB_API_KEY}
- TRADING_PROCESS_NAME=trading_system
# Resource limits
- MAX_MEMORY_GB=2.0
- LOG_LEVEL=${LOG_LEVEL:-INFO}
volumes:
# Mount data directory for persistence
- ./data:/app/data
- ./logs:/app/logs
- ./config:/app/config:ro
# Health monitoring
healthcheck:
test: ["CMD", "python", "-c", "import requests; requests.get('http://localhost:8080/health').raise_for_status()"]
interval: 30s
timeout: 10s
retries: 3
start_period: 60s
# Graceful shutdown handling
stop_grace_period: 30s
# Resource constraints
deploy:
resources:
limits:
memory: 2G
cpus: "1.0"
reservations:
memory: 512M
cpus: "0.5"
networks:
- quant_net
# Optional: Prometheus for metrics collection
prometheus:
image: prom/prometheus:latest
container_name: quant_prometheus
restart: unless-stopped
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
- prometheus_data:/prometheus
command:
- '--config.file=/etc/prometheus/prometheus.yml'
- '--storage.tsdb.path=/prometheus'
networks:
- quant_net
depends_on:
- trading_system
# Optional: Grafana for visualization
grafana:
image: grafana/grafana:latest
container_name: quant_grafana
restart: unless-stopped
environment:
- GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD:-admin}
volumes:
- grafana_data:/var/lib/grafana
ports:
- "3000:3000"
networks:
- quant_net
depends_on:
- prometheus
networks:
quant_net:
driver: bridge
volumes:
prometheus_data:
grafana_data:
GitHub Actions Deployment Pipeline
# .github/workflows/deploy.yml
# Deploys to VPS on push to main branch
name: Deploy to Production VPS
on:
push:
branches: [main]
workflow_dispatch: # Allow manual trigger
jobs:
deploy:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Login to Docker Hub
uses: docker/login-action@v3
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Build and push Docker image
uses: docker/build-push-action@v5
with:
context: .
push: true
tags: |
${{ secrets.DOCKERHUB_USERNAME }}/quant-trading:${{ github.sha }}
${{ secrets.DOCKERHUB_USERNAME }}/quant-trading:latest
cache-from: type=registry,ref=${{ secrets.DOCKERHUB_USERNAME }}/quant-trading:latest
cache-to: type=inline
- name: Deploy to VPS
uses: appleboy/ssh-action@v1.0.0
with:
host: ${{ secrets.VPS_HOST }}
username: ${{ secrets.VPS_USER }}
key: ${{ secrets.VPS_SSH_KEY }}
script: |
# Pull latest image
docker pull ${{ secrets.DOCKERHUB_USERNAME }}/quant-trading:latest
# Stop and remove old container (keep logs and data)
docker stop quant_trading || true
docker rm quant_trading || true
# Start new container with environment variables from secrets
docker run -d \
--name quant_trading \
--restart unless-stopped \
-e TICKDB_API_KEY=${{ secrets.TICKDB_API_KEY }} \
-e LOG_LEVEL=INFO \
-v $(pwd)/data:/app/data \
-v $(pwd)/logs:/app/logs \
${{ secrets.DOCKERHUB_USERNAME }}/quant-trading:latest
# Verify deployment
sleep 10
docker logs quant_trading --tail 50
docker ps | grep quant_trading
Putting It All Together: The Automated Trading Day
Here's what a fully automated trading day looks like with this stack:
| Time (ET) | Component | What happens |
|---|---|---|
| 8:30 AM | Scheduler | Pre-market data fetch job fires |
| 8:55 AM | Scheduler | Market open preparation job fires |
| 9:30 AM | Scheduler | Market open job fires — strategy begins |
| 9:30 AM - 4:00 PM | Health Monitor | Every 60s: checks disk, memory, process health |
| 9:30 AM - 4:00 PM | Logs | All strategy activity written to rotated JSON logs |
| 3:55 PM | Scheduler | EOD position close job fires |
| 4:00 PM | Scheduler | Post-market analysis job fires |
| 11:00 PM | Log rotation | Old logs compressed and shipped to S3 |
| 2:00 AM | (You sleep) | System runs without you |
The only time you need to intervene is when an alert fires — and with proper cooldown and severity routing, this should be rare.
The 80% Math
Let's quantify the "80% repetitive work" savings:
| Task | Manual time/week | Automated time/week | Savings |
|---|---|---|---|
| Starting/stopping processes | 30 min | 0 | 100% |
| Checking logs for errors | 60 min | 0 (Slack alerts instead) | 100% |
| Restarting crashed processes | 45 min | 0 (auto-restart via supervisor) | 100% |
| Deploying code updates | 30 min | 5 min (push to main) | 83% |
| Monitoring disk space | 15 min | 0 (health check) | 100% |
| Rotating old logs | 15 min | 0 (automatic) | 100% |
| Pulling market data for analysis | 60 min | 0 (scheduled jobs) | 100% |
| Total | ~4.25 hours | ~5 minutes | 98% |
The remaining 5% is the 5 minutes per week to review the automated alerts and approve deployment commits. That's not quite 80% — it's closer to 98%.
Next Steps
If you're setting up automation for the first time, start with the Health Monitor and Alert Router. They're the highest-value, lowest-effort addition to any existing system. A crashed process that wakes you up at 2 AM is infinitely better than a crashed process that loses money silently.
If you want to extend this stack:
- Add Prometheus metrics (counter, gauge, histogram) to track strategy performance
- Set up Grafana dashboards to visualize P&L, signal distribution, and system health
- Implement supervisor or systemd for process management that survives server reboots
- Add database-backed state (PostgreSQL) to track positions and orders across restarts
If you need 10+ years of historical OHLCV data for backtesting your automated strategies, the TickDB API provides clean, aligned US equity data via the /kline endpoint — suitable for cross-cycle backtesting without the data cleaning overhead.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Any automated trading system should be thoroughly tested in paper trading mode before deployment with real capital.