Building Production-Grade Data Infrastructure for a 3-Person Quant Team

When a solo quant quant becomes a quant team, the first thing that breaks is never the strategy. It is the workflow.

You have been running your mean-reversion strategy from a single laptop for eighteen months. The code lives in ~/strategies/. Your API keys are in a text file called keys_backup.txt on your desktop. Backtests run locally. When you want to test a change, you copy the script, modify it, and run it. It works. It has always worked.

Then you hire two people.

Within two weeks, you have four different versions of the same strategy running on three machines. Nobody knows which results came from which code. The production instance is running an older version than your backtester because you forgot to deploy the last update. One team member accidentally committed API keys to a shared GitHub repository. The database queries are inconsistent because each person constructed them slightly differently.

The strategy did not break. The infrastructure around the strategy broke.

This is the problem this article solves. For a three-person quantitative team, we will design a production-grade data infrastructure that handles five critical concerns: centralized data access via TickDB, Git-based code collaboration, API key security, environment configuration management, and deployment discipline. Every recommendation in this article assumes a small team with no dedicated DevOps engineer — meaning every solution must be maintainable by the quants themselves.

---

1. The Three-Layer Infrastructure Model

Before diving into implementation, we need a mental model that governs every architectural decision.

For a small quant team, infrastructure operates across three layers:

Layer 1 — Data: Where market data comes from, how it is stored, and how team members access it. TickDB serves as the canonical data source for this layer.

Layer 2 — Code: How strategy code is written, versioned, reviewed, and deployed. Git with a structured workflow governs this layer.

Layer 3 — Execution: How live and backtested strategies interact with the data layer, and how credentials flow between layers without exposure.

The critical design principle: each layer should be replaceable without breaking the others. If you switch data providers, the code layer should not need to change. If you update your deployment target, the data layer should not care. This decoupling is what a three-person team needs because you cannot afford the blast radius of a monolithic mistake.

---

2. Centralizing Data Access with TickDB

2.1 The Shared Data Problem

In a solo workflow, data access is straightforward: you call the API, you get the data, you use it. In a team workflow, this breaks in three ways.

First, latency inconsistency. If each team member calls TickDB independently from their own machines, network routing differences can cause the same query to return different results at different times. A backtest run by Person A and a live deployment run by Person B can produce divergent results not because of strategy logic but because of data timing.

Second, redundant cost. Each individual machine making independent API calls burns through rate limits faster than a centralized fetcher would. With a shared data service, one machine fetches the data, stores it locally, and serves it to the team over an internal network.

Third, schema drift. When each person writes their own data fetching code, column names diverge. Person A uses close_price. Person B uses close. Person C uses closePx. Downstream analysis becomes a cleaning nightmare.

2.2 The Canonical Fetcher Architecture

The solution is a shared data fetcher service. One machine — typically a small VPS or a dedicated development machine that stays on — runs a fetcher that pulls data from TickDB and stores it in a shared PostgreSQL database. The other two machines run strategies that query the local database, not TickDB directly.

This architecture has three components:

Fetcher service: Runs on Machine A. Pulls OHLCV data via TickDB's /v1/market/kline endpoint, pulls order book depth via the depth channel, and writes everything to a shared PostgreSQL instance.

Shared database: PostgreSQL running on Machine A (or a separate low-cost VPS). All market data lives here. Schemas are defined once and never modified — only extended.

Strategy clients: Machines B and C connect to the shared database. They never call TickDB directly. Their code receives pre-processed, consistently structured data.

This setup adds approximately 100–200ms of latency compared to direct TickDB calls, which is acceptable for backtesting and for strategies that operate on minute-scale or longer timeframes. For sub-second strategies, you would need a different architecture, but that is not the target audience of this article.

2.3 Production-Grade Fetcher Code

The fetcher service must be resilient. It runs unattended. It must handle rate limits, network interruptions, and reconnection gracefully. Below is a production-grade fetcher implementation in Python that pulls hourly OHLCV data from TickDB and writes it to PostgreSQL.

import os
import time
import json
import logging
import random
from datetime import datetime, timedelta
from typing import Optional, Dict, Any

import requests
import psycopg2
from psycopg2.extras import execute_batch

# Configure logging for production monitoring
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("/var/log/tickdb-fetcher/fetcher.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)


class TickDBFetcher:
    """
    Production-grade TickDB data fetcher with:
    - Exponential backoff + jitter on reconnect
    - Rate limit handling (code 3001)
    - PostgreSQL bulk insert with conflict resolution
    - Configurable via environment variables
    - Structured logging for monitoring
    """
    
    def __init__(self):
        self.api_key = os.environ.get("TICKDB_API_KEY")
        if not self.api_key:
            raise ValueError("TICKDB_API_KEY environment variable is required")
        
        self.base_url = "https://api.tickdb.ai/v1"
        self.headers = {"X-API-Key": self.api_key}
        
        # PostgreSQL connection via environment variables
        self.db_config = {
            "host": os.environ.get("PGHOST", "localhost"),
            "port": os.environ.get("PGPORT", "5432"),
            "database": os.environ.get("PGDATABASE", "market_data"),
            "user": os.environ.get("PGUSER", "quant_team"),
            "password": os.environ.get("PGPASSWORD"),
        }
        
        self.db_conn = None
        self.max_retries = 5
        self.base_delay = 2.0  # seconds
    
    def connect_db(self) -> None:
        """Establish PostgreSQL connection with retry logic."""
        for attempt in range(3):
            try:
                self.db_conn = psycopg2.connect(**self.db_config)
                self.db_conn.autocommit = False
                logger.info("PostgreSQL connection established")
                return
            except psycopg2.OperationalError as e:
                logger.warning(f"Database connection attempt {attempt + 1} failed: {e}")
                time.sleep(5 * (attempt + 1))
        raise RuntimeError("Could not establish database connection after 3 attempts")
    
    def _request_with_retry(
        self,
        method: str,
        endpoint: str,
        params: Optional[Dict[str, Any]] = None,
        retries: int = 0
    ) -> Dict[str, Any]:
        """
        HTTP request with exponential backoff + jitter and rate limit handling.
        
        Handles:
        - Network errors: retry with exponential backoff
        - Rate limits (3001): read Retry-After header and wait
        - Authentication errors (1001/1002): fail immediately
        - Server errors (5xx): retry up to max_retries
        """
        url = f"{self.base_url}{endpoint}"
        
        try:
            response = requests.request(
                method,
                url,
                headers=self.headers,
                params=params,
                timeout=(3.05, 15)  # (connect timeout, read timeout)
            )
            
            # Parse response
            try:
                data = response.json()
            except json.JSONDecodeError:
                logger.error(f"Non-JSON response from {endpoint}: {response.text[:200]}")
                raise RuntimeError(f"Invalid JSON from API: {endpoint}")
            
            # Handle error codes
            code = data.get("code", 0)
            if code == 0:
                return data
            
            if code in (1001, 1002):
                # Authentication errors — do not retry
                raise ValueError(
                    f"Authentication failed (code {code}). "
                    "Verify TICKDB_API_KEY environment variable."
                )
            
            if code == 3001:
                # Rate limit — extract Retry-After
                retry_after = int(response.headers.get("Retry-After", 60))
                logger.warning(f"Rate limited. Waiting {retry_after}s before retry.")
                time.sleep(retry_after)
                return self._request_with_retry(method, endpoint, params, retries)
            
            if code == 2002:
                # Symbol not found — this is expected for some symbols; log and skip
                logger.warning(f"Symbol not found: {params.get('symbol', 'unknown')}")
                return {"data": None}
            
            # Server errors — retry with backoff
            if retries < self.max_retries:
                delay = self.base_delay * (2 ** retries)
                jitter = random.uniform(0, delay * 0.1)
                logger.warning(
                    f"Server error {code} at {endpoint}. "
                    f"Retrying in {delay + jitter:.1f}s (attempt {retries + 1}/{self.max_retries})"
                )
                time.sleep(delay + jitter)
                return self._request_with_retry(method, endpoint, params, retries + 1)
            
            raise RuntimeError(f"Unrecoverable error {code}: {data.get('message', 'Unknown')}")
            
        except requests.exceptions.Timeout:
            if retries < self.max_retries:
                delay = self.base_delay * (2 ** retries)
                logger.warning(f"Request timeout. Retrying in {delay}s (attempt {retries + 1})")
                time.sleep(delay)
                return self._request_with_retry(method, endpoint, params, retries + 1)
            raise
        
        except requests.exceptions.RequestException as e:
            logger.error(f"Network error during request: {e}")
            raise
    
    def get_kline(
        self,
        symbol: str,
        interval: str = "1h",
        start_time: Optional[int] = None,
        end_time: Optional[int] = None,
        limit: int = 500
    ) -> list:
        """Fetch OHLCV kline data for a given symbol."""
        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": limit,
        }
        
        if start_time:
            params["start_time"] = start_time
        if end_time:
            params["end_time"] = end_time
        
        result = self._request_with_retry("GET", "/market/kline", params=params)
        return result.get("data", []) or []
    
    def upsert_kline(self, symbol: str, interval: str, candles: list) -> int:
        """
        Upsert kline data into PostgreSQL using INSERT ... ON CONFLICT.
        Returns the number of rows inserted/updated.
        """
        if not candles:
            return 0
        
        query = """
        INSERT INTO ohlcv (symbol, interval, open_time, open, high, low, close, volume, close_time)
        VALUES (%(symbol)s, %(interval)s, %(open_time)s, %(open)s, %(high)s, %(low)s, %(close)s, %(volume)s, %(close_time)s)
        ON CONFLICT (symbol, interval, open_time) 
        DO UPDATE SET
            open = EXCLUDED.open,
            high = EXCLUDED.high,
            low = EXCLUDED.low,
            close = EXCLUDED.close,
            volume = EXCLUDED.volume,
            close_time = EXCLUDED.close_time,
            updated_at = NOW()
        """
        
        records = [
            {
                "symbol": c.get("symbol", symbol),
                "interval": c.get("interval", interval),
                "open_time": c.get("open_time"),
                "open": float(c.get("open", 0)),
                "high": float(c.get("high", 0)),
                "low": float(c.get("low", 0)),
                "close": float(c.get("close", 0)),
                "volume": float(c.get("volume", 0)),
                "close_time": c.get("close_time"),
            }
            for c in candles
        ]
        
        with self.db_conn.cursor() as cur:
            execute_batch(cur, query, records)
        self.db_conn.commit()
        
        return len(records)
    
    def fetch_and_store(
        self,
        symbols: list,
        interval: str = "1h",
        lookback_days: int = 30
    ) -> Dict[str, int]:
        """
        Main fetch-and-store routine. Fetches recent kline data for all symbols
        and upserts into PostgreSQL.
        
        Returns a dict mapping symbol -> rows affected.
        """
        results = {}
        end_time = int(datetime.now().timestamp() * 1000)
        start_time = int((datetime.now() - timedelta(days=lookback_days)).timestamp() * 1000)
        
        for symbol in symbols:
            try:
                logger.info(f"Fetching {symbol} ({interval}) from {start_time} to {end_time}")
                candles = self.get_kline(symbol, interval, start_time, end_time)
                rows = self.upsert_kline(symbol, interval, candles)
                results[symbol] = rows
                logger.info(f"{symbol}: upserted {rows} candles")
                
                # Respect rate limits between symbols
                time.sleep(0.5)
                
            except Exception as e:
                logger.error(f"Failed to fetch {symbol}: {e}")
                results[symbol] = 0
        
        return results
    
    def run_continuous(
        self,
        symbols: list,
        interval: str = "1h",
        fetch_interval: int = 3600
    ):
        """
        Continuous fetcher loop. Designed to run as a systemd service.
        Fetches latest data every fetch_interval seconds.
        """
        logger.info(f"Starting continuous fetcher for symbols: {symbols}")
        
        while True:
            try:
                if not self.db_conn or self.db_conn.closed:
                    self.connect_db()
                
                self.fetch_and_store(symbols, interval, lookback_days=2)
                logger.info(f"Sleeping for {fetch_interval}s")
                
            except Exception as e:
                logger.error(f"Continuous fetcher error: {e}")
                # Longer sleep on error to prevent rapid retry loops
                time.sleep(60)
            
            time.sleep(fetch_interval)


if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="TickDB to PostgreSQL fetcher")
    parser.add_argument(
        "--symbols",
        nargs="+",
        default=["AAPL.US", "NVDA.US", "TSLA.US"],
        help="Symbols to fetch"
    )
    parser.add_argument(
        "--interval",
        default="1h",
        choices=["1m", "5m", "15m", "1h", "4h", "1d"],
        help="Kline interval"
    )
    parser.add_argument(
        "--continuous",
        action="store_true",
        help="Run continuously (for production deployment)"
    )
    
    args = parser.parse_args()
    
    fetcher = TickDBFetcher()
    fetcher.connect_db()
    
    if args.continuous:
        fetcher.run_continuous(args.symbols, args.interval)
    else:
        results = fetcher.fetch_and_store(args.symbols, args.interval)
        for symbol, count in results.items():
            print(f"{symbol}: {count} rows")

Critical engineering notes for the above code:

1. Heartbeat is implicit via the continuous loop. The _request_with_retry method handles connection keepalive through its timeout configuration.
2. Exponential backoff + jitter prevents thundering herd when the API recovers from an outage. Three team members running the same fetcher simultaneously will not synchronized their retries.
3. Rate limit handling checks for code 3001 and reads the Retry-After header. This is not optional — the TickDB API enforces rate limits and ignoring them will result in temporary IP bans.
4. Database upsert pattern using ON CONFLICT means the same fetcher can run on any schedule without creating duplicate rows. The primary key is (symbol, interval, open_time).

2.4 PostgreSQL Schema for Shared Market Data

The shared database schema must be stable and well-documented. Here is the base schema that supports OHLCV storage with an audit trail:

-- Market data schema
CREATE TABLE IF NOT EXISTS ohlcv (
    symbol      VARCHAR(20) NOT NULL,
    interval    VARCHAR(10) NOT NULL,
    open_time   BIGINT NOT NULL,  -- milliseconds since epoch
    open        DECIMAL(18, 8) NOT NULL,
    high        DECIMAL(18, 8) NOT NULL,
    low         DECIMAL(18, 8) NOT NULL,
    close       DECIMAL(18, 8) NOT NULL,
    volume      DECIMAL(18, 8) NOT NULL,
    close_time  BIGINT,
    created_at  TIMESTAMP DEFAULT NOW(),
    updated_at  TIMESTAMP DEFAULT NOW(),
    
    PRIMARY KEY (symbol, interval, open_time)
);

-- Indexes for common query patterns
CREATE INDEX IF NOT EXISTS idx_ohlcv_symbol_interval_time
ON ohlcv (symbol, interval, open_time DESC);

CREATE INDEX IF NOT EXISTS idx_ohlcv_updated
ON ohlcv (updated_at DESC);

-- Audit log for tracking data freshness
CREATE TABLE IF NOT EXISTS fetch_audit (
    id          SERIAL PRIMARY KEY,
    symbol      VARCHAR(20) NOT NULL,
    interval    VARCHAR(10) NOT NULL,
    rows_fetched INT NOT NULL,
    fetch_start  TIMESTAMP NOT NULL,
    fetch_end    TIMESTAMP NOT NULL,
    status       VARCHAR(20) NOT NULL,  -- 'success', 'partial', 'failed'
    error_msg    TEXT
);

-- Function to log fetch results
CREATE OR REPLACE FUNCTION log_fetch(
    p_symbol VARCHAR,
    p_interval VARCHAR,
    p_rows INT,
    p_start TIMESTAMP,
    p_end TIMESTAMP,
    p_status VARCHAR,
    p_error TEXT DEFAULT NULL
) RETURNS VOID AS $$
BEGIN
    INSERT INTO fetch_audit (symbol, interval, rows_fetched, fetch_start, fetch_end, status, error_msg)
    VALUES (p_symbol, p_interval, p_rows, p_start, p_end, p_status, p_error);
END;
$$ LANGUAGE plpgsql;

This schema solves the schema drift problem. Every team member queries the same table with the same column names. The fetch_audit table provides observability — if a strategy produces unexpected results, you can check whether the data was actually fresh when it ran.

---

3. Git-Based Code Collaboration

3.1 Why Git Is Non-Negotiable for Three-Person Teams

Git is not optional for a quant team, even a small one. The reason is not version control — it is the ability to answer one specific question with certainty: "What code produced these results?"

In a backtest-driven strategy development process, you will run hundreds of experiments. You will tweak parameters, refactor signal calculations, and test different entry conditions. Without Git, each experiment is a file copy with a slightly different name. When a strategy performs well in backtest and then fails in live trading, you have no reliable way to identify what changed.

With Git, every result is tied to a specific commit hash. You can reconstruct exactly what the strategy was doing when it generated those numbers.

3.2 The Three-Branch Workflow

For a three-person quant team, we recommend a simplified Git workflow with three persistent branches and one transient branch pattern:

Main branches:

• main: Production code only. Live strategies run from main. Only deploys via pull request after review.
• develop: Integration branch. All feature branches merge here before review. Code here has been tested locally but not peer-reviewed.
• staging: Mirrors production data environment. Strategies are backtested against staging before deployment to main. This branch exists to catch data access bugs before they hit production.

Transient pattern:

• feature/your-name-description: Created from develop. One branch per task. Rebased onto develop before merge.

Merge workflow:

1. Person A completes a feature on feature/a-momentum-signal.
2. Person A opens a pull request from feature/a-momentum-signal to develop.
3. Person B reviews the PR — checks strategy logic, verifies backtest improvement, reviews code style.
4. After approval, feature/a-momentum-signal is merged into develop.
5. On merge to develop, a CI pipeline (even a simple shell script) runs the backtest and logs results to the fetch_audit table.
6. Weekly, develop is merged into staging and backtests are run against production-equivalent data.
7. Monthly or when staging is stable, staging merges into main for production deployment.

This workflow creates three checkpoints where code quality is verified: feature merge, develop-to-staging, and staging-to-main.

3.3 Repository Structure

A well-organized repository structure is as important as the workflow itself. Here is a recommended structure for a three-person quant team's codebase:

quant-team/
├── README.md                    # Project overview, setup instructions
├── .env.example                 # Template for environment variables (NO real keys)
├── .gitignore                   # Excludes .env, __pycache__, *.pyc, .pytest_cache
├── docker/
│   ├── Dockerfile.fetcher       # Data fetcher container
│   ├── Dockerfile.strategy      # Strategy execution container
│   └── docker-compose.yml       # Local development environment
├── db/
│   ├── schema.sql               # PostgreSQL schema
│   └── migrations/              # Versioned schema changes
├── src/
│   ├── data/                    # Data fetching and processing
│   │   ├── fetcher.py
│   │   ├── database.py
│   │   └── schemas.py
│   ├── strategies/              # One subdirectory per strategy
│   │   ├── momentum/
│   │   │   ├── __init__.py
│   │   │   ├── signals.py
│   │   │   ├── backtest.py
│   │   │   └── live.py
│   │   └── mean_reversion/
│   │       ├── __init__.py
│   │       ├── signals.py
│   │       ├── backtest.py
│   │       └── live.py
│   ├── common/                  # Shared utilities
│   │   ├── config.py            # Environment variable loader
│   │   ├── logging_config.py
│   │   ├── risk_limits.py
│   │   └── performance.py        # Performance measurement
│   └── tests/
│       ├── test_signals.py
│       ├── test_backtest.py
│       └── test_integration.py
├── notebooks/                   # Exploratory analysis (not strategy source)
│   └── exploratory/
│       └── spread_analysis.ipynb
├── backtests/                   # Output directory for backtest results
│   └── results/
└── docs/
    ├── architecture.md
    ├── data_dictionary.md
    └── deployment.md

Key structural principles:

1. Strategy isolation: Each strategy lives in its own subdirectory. A change to the mean-reversion strategy cannot accidentally break momentum without a merge conflict.
2. Shared common library: Data access, logging, and configuration utilities live in one place. When the database schema changes, you update db/schema.sql and src/data/database.py in one PR.
3. backtests/ as output only: Backtest results are generated artifacts. They are committed to Git so that you can compare results across commits, but they are never edited manually.
4. notebooks/ is exploratory: Jupyter notebooks are for research, not production code. They are not subject to the same review rigor as strategy code.

3.4 The Critical .gitignore for Quant Teams

One of the most common security incidents for small quant teams is accidentally committing API keys to a public or shared repository. Your .gitignore must include these entries at minimum:

# Environment and secrets
.env
.env.local
.env.*.local
*.pem
*.key
keys.txt
secrets.yml

# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
dist/
*.egg-info/
.eggs/

# Data files (these should come from the shared database, not the repo)
*.csv
*.parquet
*.h5

# IDE
.vscode/
.idea/
*.swp
*.swo

# Jupyter
.ipynb_checkpoints/
*/.ipynb_checkpoints/*

# Testing
.pytest_cache/
.coverage
htmlcov/

# OS
.DS_Store
Thumbs.db

And your .env.example file should look like this — never containing real keys:

# TickDB API Configuration
TICKDB_API_KEY=your_api_key_here

# PostgreSQL Configuration (for fetcher service)
PGHOST=localhost
PGPORT=5432
PGDATABASE=market_data
PGUSER=quant_team
PGPASSWORD=your_local_password_here

# Strategy Configuration
MAX_POSITION_SIZE=10000
RISK_LIMIT_DAILY=5000
LOG_LEVEL=INFO

# Deployment
DEPLOYMENT_ENV=development  # development, staging, production

---

4. API Key Management in a Team Context

4.1 The Problem with Shared Keys

The naive approach to API key management for a small team is to share one API key and store it in a shared text file or a team chat message. This approach has three critical failure modes:

Exposure risk: The moment a key is sent over Slack or email, it exists in multiple places: the sender's outbox, the recipient's inbox, and potentially server logs. If any of those systems is compromised, the key is compromised.

Audit blindness: If you use a shared key, you cannot determine which team member made a particular API call. When something goes wrong — a rate limit is hit, a suspicious query is made, a bill is higher than expected — you have no attribution.

Credential rotation paralysis: When a key needs to be rotated (after a team member leaves, after a suspected compromise, or as part of routine security hygiene), you must update every place the key was shared. For a small team that has been sharing keys casually, this can mean hunting through dozens of files and messages.

4.2 The Environment Variable + Vault Approach

For a three-person team without a dedicated DevOps engineer, the practical solution is a combination of environment variables and a lightweight secret manager.

Step 1: Individual keys. Each team member generates their own API key from the TickDB dashboard. Each person stores their own key in their local .env file, which is never committed to Git. This ensures attribution — you can see exactly which key made which call in the TickDB dashboard.

Step 2: The fetcher's key. The shared data fetcher service (Machine A) needs its own dedicated key. This key is stored in an environment variable on Machine A only, set via the hosting provider's secret management (e.g., AWS Secrets Manager, DigitalOcean secrets, or a simple systemd environment file with restricted permissions).

Step 3: Local development. For local development, each person copies .env.example to .env, fills in their individual key, and runs the code. The codebase never contains a real key — only the template.

# Set up local environment (run once per machine)
cp .env.example .env
# Edit .env with your personal API key
chmod 600 .env  # Restrict file permissions

Step 4: Production deployment. The fetcher service reads its key from a systemd environment file:

# /etc/systemd/system/tickdb-fetcher.service.d/environment.conf
[Service]
Environment="TICKDB_API_KEY=tk_live_your_production_key_here"
Environment="PGPASSWORD=your_db_password_here"

# /etc/systemd/system/tickdb-fetcher.service
[Unit]
Description=TickDB Data Fetcher
After=network.target postgresql.service

[Service]
Type=simple
User=tickdb-fetcher
WorkingDirectory=/opt/tickdb-fetcher
ExecStart=/usr/bin/python3 /opt/tickdb-fetcher/fetcher.py --continuous --symbols AAPL.US NVDA.US TSLA.US --interval 1h
Restart=on-failure
RestartSec=10s

# Run as a non-root user with limited permissions
User=tickdb-fetcher
Group=tickdb-fetcher

[Install]
WantedBy=multi-user.target

This approach means that even if a team member's laptop is compromised, the production key on Machine A is unaffected. And if a team member leaves, you revoke their individual key — the fetcher keeps running.

4.3 Key Rotation Schedule

For a quant team, we recommend the following rotation schedule:

Key type	Rotation frequency	Trigger
Individual developer keys	Every 6 months, or immediately on departure	Calendar or HR event
Production fetcher key	Every 12 months	Calendar reminder
Database password	Every 12 months	Calendar reminder
All keys	Immediately	Any suspected compromise

TickDB's key management dashboard makes rotation straightforward: generate a new key, update the environment variable, restart the service, verify data flow, then revoke the old key.

---

5. Deployment Discipline: From Backtest to Live

5.1 The Deployment Pipeline

The transition from backtest to live is where most small quant teams get sloppy. The pressure to deploy a working strategy overrides discipline, and shortcuts accumulate. Here is a structured pipeline that three people can actually follow:

Stage 1 — Development:

• Strategy runs on a local machine.
• Backtested against the shared PostgreSQL database.
• Results committed to backtests/results/ with a descriptive name and a reference to the commit hash.
• Code lives in a feature/ branch.

Stage 2 — Review:

• Pull request opened to develop.
• Second team member reviews the strategy logic and backtest assumptions.
• Review checklist: Is slippage modeled correctly? Are risk limits enforced? Is the backtest period sufficient (at least 3 years covering one bull-bear cycle)?

Stage 3 — Staging:

• Strategy deployed to a staging environment that connects to the production database.
• Paper trading mode enabled (signals generated but orders not executed).
• Run for a minimum of 5 trading days.
• Compare paper trading results against the backtest for the same period. Divergence > 10% triggers a review.

Stage 4 — Production:

• Strategy deployed to production environment.
• Hard risk limits enforced at the execution layer — not just in the strategy code.
• Monitoring alerts configured for: drawdown exceeding threshold, position size exceeding limit, data feed interruption.
• Live trading results compared against paper trading weekly.

5.2 Infrastructure Configuration by Team Size

Given that your team is three people, here is a practical infrastructure recommendation across three tiers:

Component	Minimum (lowest cost)	Recommended	Scalable
Data fetcher host	Raspberry Pi 4 ($50) or a $10/mo VPS	$20–30/mo VPS (DigitalOcean, Linode)	Dedicated $100+/mo VPS
Shared database	PostgreSQL on fetcher host	Separate $15/mo managed PostgreSQL	RDS or Cloud SQL with replica
Strategy execution	Developer laptops	Dedicated $20/mo VPS per strategy	Kubernetes cluster
Monitoring	Paper logging	Uptime monitoring + Slack alerts	Grafana + Prometheus
Backup	Manual weekly pg_dump	Automated daily backups	Continuous WAL archiving

For most three-person teams starting out, the "Recommended" column is the sweet spot. The "Minimum" column works for proof-of-concept but introduces fragility — a Raspberry Pi going offline at 2 AM during an earnings report is not a good experience.

5.3 The Non-Negotiable Risk Limit Implementation

Every live strategy must have hard risk limits enforced at the execution layer. This is not optional, and it must survive a bug in the strategy code itself. The enforcement layer sits between the strategy signal and the order submission:

import os
from decimal import Decimal
from typing import Optional
from dataclasses import dataclass
from datetime import datetime


@dataclass
class RiskLimits:
    """Risk limits loaded from environment variables."""
    max_position_size: Decimal = Decimal(os.environ.get("MAX_POSITION_SIZE", "10000"))
    max_daily_loss: Decimal = Decimal(os.environ.get("MAX_DAILY_LOSS", "5000"))
    max_single_order: Decimal = Decimal(os.environ.get("MAX_SINGLE_ORDER", "2000"))
    max_orders_per_day: int = int(os.environ.get("MAX_ORDERS_PER_DAY", "50"))


@dataclass
class OrderRequest:
    """A validated order request."""
    symbol: str
    side: str          # 'buy' or 'sell'
    quantity: Decimal
    price: Optional[Decimal] = None


class RiskEnforcer:
    """
    Hard risk limit enforcement layer.
    
    This class enforces risk limits BEFORE any order is submitted.
    It is independent of the strategy code — a bug in the strategy
    cannot bypass these limits.
    """
    
    def __init__(self, limits: RiskLimits):
        self.limits = limits
        self.daily_pnl = Decimal("0")
        self.orders_today = 0
        self.last_reset = datetime.now().date()
    
    def _check_daily_reset(self) -> None:
        """Reset daily counters at market open (9:30 AM ET)."""
        today = datetime.now().date()
        if today > self.last_reset:
            self.daily_pnl = Decimal("0")
            self.orders_today = 0
            self.last_reset = today
    
    def validate_order(
        self,
        order: OrderRequest,
        current_position: Decimal,
        current_price: Decimal
    ) -> tuple[bool, Optional[str]]:
        """
        Validate an order against all risk limits.
        
        Returns (is_valid, rejection_reason).
        If is_valid is False, rejection_reason contains the specific violation.
        """
        self._check_daily_reset()
        
        # Calculate projected position
        if order.side == "buy":
            projected_position = current_position + order.quantity
        else:
            projected_position = current_position - order.quantity
        
        projected_cost = projected_position * current_price
        
        # Check 1: Max position size
        if projected_cost > self.limits.max_position_size:
            return False, (
                f"Position size limit exceeded. "
                f"Projected: ${projected_cost}, Limit: ${self.limits.max_position_size}"
            )
        
        # Check 2: Single order size
        order_value = order.quantity * (order.price or current_price)
        if order_value > self.limits.max_single_order:
            return False, (
                f"Single order limit exceeded. "
                f"Order value: ${order_value}, Limit: ${self.limits.max_single_order}"
            )
        
        # Check 3: Daily order count
        if self.orders_today >= self.limits.max_orders_per_day:
            return False, (
                f"Daily order limit reached. "
                f"Orders today: {self.orders_today}, Limit: {self.limits.max_orders_per_day}"
            )
        
        # Check 4: Daily loss limit (only checked before submitting a new order)
        # This is a soft check — the strategy should manage this, but we enforce it here
        # by blocking orders when we're already at the daily loss limit
        if self.daily_pnl < -self.limits.max_daily_loss:
            return False, (
                f"Daily loss limit exceeded. "
                f"Daily P&L: ${self.daily_pnl}, Limit: -${self.limits.max_daily_loss}. "
                f"No new orders will be submitted today."
            )
        
        return True, None
    
    def record_trade(self, symbol: str, side: str, quantity: Decimal, price: Decimal, pnl: Decimal = Decimal("0")) -> None:
        """Record a completed trade for limit tracking."""
        self._check_daily_reset()
        self.orders_today += 1
        self.daily_pnl += pnl
    
    def get_status(self) -> dict:
        """Return current risk status for monitoring."""
        self._check_daily_reset()
        return {
            "daily_pnl": float(self.daily_pnl),
            "daily_loss_remaining": float(self.limits.max_daily_loss + self.daily_pnl),
            "orders_today": self.orders_today,
            "orders_remaining": self.limits.max_orders_per_day - self.orders_today,
            "position_limit_remaining": float(self.limits.max_position_size),
        }

This RiskEnforcer class is intentionally simple and independent. It does not know about your strategy's logic. It only knows about money. A strategy bug that generates a bad signal cannot bypass these limits because the strategy code calls risk_enforcer.validate_order() before submitting anything to the exchange.

---

6. Deployment Tier Recommendations by Team Stage

Not every three-person team needs the same infrastructure. Here is a practical guide matching infrastructure investment to team maturity:

Team stage	Description	Recommended setup
Stage 1: Proof of concept	Testing strategy ideas; no live capital	Shared PostgreSQL on fetcher VPS; strategies on individual laptops; manual backtests
Stage 2: Active development	Multiple strategies in development; weekly backtests	Dedicated fetcher VPS with managed PostgreSQL; staging environment on second VPS; Git workflow fully established
Stage 3: Live trading	One or more strategies running with real capital	Production fetcher + staging separate; risk enforcement layer deployed; monitoring alerts active
Stage 4: Scaling	Multiple strategies; considering additional team members	Infrastructure-as-code (Terraform/Ansible); separate execution environments per strategy; formal incident response

Most three-person teams start in Stage 1 and move to Stage 3 over 6–12 months. The jump from Stage 3 to Stage 4 typically coincides with either a capital increase or a fourth team member joining — at which point the infrastructure described in this article begins to strain at its seams and a more formal DevOps practice becomes necessary.

---

7. Putting It All Together

The five concerns we addressed — centralized data access, Git-based collaboration, API key management, environment configuration, and deployment discipline — are not independent problems. They form a system.

When a team member joins:

1. They clone the repository.
2. They copy .env.example to .env and add their individual TickDB API key.
3. They run psql against the shared database to access historical data.
4. They create a feature/ branch, implement a strategy change, and open a pull request.
5. After review, the change is merged to develop, deployed to staging, and eventually to production.

This workflow is self-documenting, auditable, and resilient to team churn. Each piece — the fetcher, the database schema, the Git history, the environment configuration — is designed to be maintained by the quants themselves without requiring a dedicated DevOps hire.

The underlying principle is this: infrastructure for a three-person team must be maintainable by generalists. Every recommendation in this article can be understood, implemented, debugged, and modified by a quant who has read this far. That is the design goal.

---

Next Steps

If you are a solo quant transitioning to a team, start with the shared data fetcher and the Git workflow. These two changes alone will eliminate the most common sources of inconsistency and error.

If your team is already sharing code, audit your .gitignore immediately. Verify that no API keys have been committed. If they have, rotate the keys immediately and use Git history rewriting tools to remove them from the repository.

If you need 10+ years of historical OHLCV data for strategy backtesting across multiple markets, TickDB provides cleaned and aligned US equity data via its /v1/market/kline endpoint. Individual API keys can be generated per team member from the dashboard, enabling the attribution-based key management described in this article. Reach out to enterprise@tickdb.ai for institutional plans covering additional asset classes.

If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace to streamline data fetching integration into your strategy code.

This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Production trading infrastructure requires careful testing and risk management.