For most retail traders, market close means the trading day is over. Close the platform, check positions, done.

For professional quant teams, close is when the real work begins.

Between 4:00 PM ET and midnight, institutional systematic desks execute a tightly choreographed sequence: archive the day's tick data, compute factor exposures, run performance attribution, identify regime changes, pre-compute tomorrow's signal candidates, and push updated models to production. The teams that do this manually spend 3–5 hours per night on repetitive tasks. The teams that automate it spend 20 minutes on exception handling.

This article dissects the architecture of a production-grade post-market workflow — from data archival to strategy attribution to next-day signal pre-computation. Every code module is production-ready: authentication via environment variables, error handling with exponential backoff, and orchestrated scheduling that survives a server restart.

The Anatomy of a Post-Market Workflow

A quant team's post-market pipeline typically operates in three sequential phases, each with distinct data dependencies and computational requirements.

Phase 1: Data Archival and Quality Control (4:00 PM – 4:45 PM ET)

The trading day generates raw data across multiple sources: order fills, position updates, market data snapshots, and execution quality metrics. This data must be archived in a format suitable for future backtesting and joined across sources by timestamp.

Phase 2: Strategy Attribution and Risk Review (4:45 PM – 8:00 PM ET)

Once data is archived and cleaned, the team decomposes performance into its constituent parts. Which factors drove returns? Did execution quality degrade in any asset class? Did any position breach intraday risk limits without triggering an alert? This phase produces the inputs for the next morning's risk meeting.

Phase 3: Signal Pre-Computation and Model Refresh (8:00 PM – midnight ET)

The final phase is forward-looking. The system pre-computes signal candidates across the next day's watchlist, runs mean-reversion scans on the overnight session, checks for earnings announcements and macro data releases, and pushes updated model weights to production. By 6:00 AM, the trading system has a head start.

Phase 1: Data Archival with TickDB

The foundation of any post-market workflow is reliable data archival. TickDB's kline endpoint provides clean, timestamp-aligned OHLCV data suitable for backtesting and strategy analytics. The critical design decision is whether to archive data as it arrives during the session or to pull the completed candle from the API after market close.

For US equity strategies, the latter approach is cleaner: the completed 1-minute or 5-minute candle has been formed and is less likely to require correction. However, for real-time dashboards during the session, the kline/latest endpoint provides the in-progress candle.

The following Python module implements a data archival pipeline that:

  • Pulls completed klines for a configurable list of symbols
  • Handles rate limits with exponential backoff
  • Writes to a local Parquet archive with partition by date and symbol
  • Validates data completeness (checks for missing bars)
import os
import time
import logging
from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd
import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────────────────────
# Configuration
# ─────────────────────────────────────────────────────────────────────────────
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
if not TICKDB_API_KEY:
    raise ValueError("TICKDB_API_KEY environment variable is required")

BASE_URL = "https://api.tickdb.ai/v1"
ARCHIVE_ROOT = Path(os.environ.get("TICKDB_ARCHIVE_ROOT", "./data/archive"))

WATCHLIST = [
    "AAPL.US", "MSFT.US", "NVDA.US", "SPY.US", "QQQ.US",
    "TSLA.US", "META.US", "AMZN.US", "AMD.US", "JPM.US"
]

INTERVAL = "5m"   # 5-minute bars
LOOKBACK_BARS = 80  # Last 80 bars of the session (~6.5 hours incl. pre/post)


# ─────────────────────────────────────────────────────────────────────────────
# API Client with rate-limit handling
# ─────────────────────────────────────────────────────────────────────────────
class TickDBClient:
    """Production-grade TickDB API client with retry logic."""

    def __init__(self, api_key: str, base_url: str = BASE_URL):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({"X-API-Key": api_key})

    @retry(
        retry=retry_if_exception_type((requests.exceptions.RequestException, RuntimeError)),
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=1, min=2, max=30)
    )
    def get_kline(self, symbol: str, interval: str, limit: int, end_time: int = None):
        """
        Fetch OHLCV klines from TickDB.

        Args:
            symbol: Trading symbol, e.g. "AAPL.US"
            interval: Kline interval, e.g. "5m", "1h"
            limit: Number of bars to fetch
            end_time: Unix timestamp in milliseconds (optional, defaults to now)

        Returns:
            List of kline dicts or raises on error.
        """
        params = {"symbol": symbol, "interval": interval, "limit": limit}
        if end_time:
            params["end_time"] = end_time

        response = self.session.get(
            f"{self.base_url}/market/kline",
            params=params,
            timeout=(3.05, 10)
        )

        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 5))
            logger.warning(f"Rate limited. Sleeping {retry_after}s")
            time.sleep(retry_after)
            raise RuntimeError("Rate limit exceeded")

        data = response.json()
        code = data.get("code", 0)

        if code == 0:
            return data.get("data", [])
        if code in (1001, 1002):
            raise ValueError(f"Invalid API key — check TICKDB_API_KEY env var")
        if code == 2002:
            raise KeyError(f"Symbol {symbol} not found")
        if code == 3001:
            retry_after = int(data.get("retry_after", 5))
            logger.warning(f"Server-side rate limit. Retrying after {retry_after}s")
            time.sleep(retry_after)
            raise RuntimeError("Server rate limit")
        raise RuntimeError(f"Unexpected error {code}: {data.get('message')}")


# ─────────────────────────────────────────────────────────────────────────────
# Data Quality Validation
# ─────────────────────────────────────────────────────────────────────────────
def validate_completeness(df: pd.DataFrame, expected_bars: int, symbol: str) -> bool:
    """
    Check whether the fetched kline data is complete.

    A complete session should have exactly expected_bars bars.
    Fewer bars indicate data gaps that need investigation.
    """
    actual_bars = len(df)
    if actual_bars < expected_bars * 0.95:  # 5% tolerance for brief exchange gaps
        logger.warning(
            f"[{symbol}] Data gap detected: expected ~{expected_bars}, got {actual_bars} bars. "
            f"Missing {expected_bars - actual_bars} bars."
        )
        return False
    logger.info(f"[{symbol}] Data completeness: {actual_bars}/{expected_bars} bars ✓")
    return True


# ─────────────────────────────────────────────────────────────────────────────
# Archive Writer
# ─────────────────────────────────────────────────────────────────────────────
def write_parquet(df: pd.DataFrame, symbol: str, trade_date: str):
    """
    Write kline data to a partitioned Parquet archive.

    Partition structure: {ARCHIVE_ROOT}/{trade_date}/{symbol}.parquet
    """
    date_path = ARCHIVE_ROOT / trade_date
    date_path.mkdir(parents=True, exist_ok=True)

    output_path = date_path / f"{symbol.replace('.', '_')}.parquet"
    df.to_parquet(output_path, index=False)
    logger.info(f"Archived {len(df)} bars for {symbol} → {output_path}")


# ─────────────────────────────────────────────────────────────────────────────
# Main Archival Pipeline
# ─────────────────────────────────────────────────────────────────────────────
def archive_session(client: TickDBClient, trade_date: str):
    """
    Main entry point: archive last session's klines for all symbols in WATCHLIST.
    """
    today = datetime.strptime(trade_date, "%Y-%m-%d")
    end_time_ms = int(today.timestamp() * 1000) + (16 * 3600 * 1000)  # 4:00 PM ET

    results = []
    for symbol in WATCHLIST:
        try:
            raw = client.get_kline(
                symbol=symbol,
                interval=INTERVAL,
                limit=LOOKBACK_BARS,
                end_time=end_time_ms
            )

            if not raw:
                logger.warning(f"[{symbol}] No data returned — check symbol or session time")
                continue

            df = pd.DataFrame(raw)
            df["symbol"] = symbol
            df["trade_date"] = trade_date

            # Ensure numeric types for analytics
            numeric_cols = ["open", "high", "low", "close", "volume"]
            for col in numeric_cols:
                df[col] = pd.to_numeric(df[col], errors="coerce")

            validate_completeness(df, LOOKBACK_BARS, symbol)
            write_parquet(df, symbol, trade_date)
            results.append({"symbol": symbol, "bars": len(df), "status": "success"})

        except Exception as exc:
            logger.error(f"[{symbol}] Archival failed: {exc}")
            results.append({"symbol": symbol, "status": "error", "error": str(exc)})

    success_count = sum(1 for r in results if r["status"] == "success")
    logger.info(f"Archival complete: {success_count}/{len(WATCHLIST)} symbols archived")

    return results


if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("Usage: python archive_pipeline.py 2026-04-15")
        sys.exit(1)

    trade_date = sys.argv[1]
    client = TickDBClient(TICKDB_API_KEY)
    archive_session(client, trade_date)

Engineering notes:

  • The tenacity library handles retry logic declaratively. Exponential backoff with jitter prevents thundering-herd behavior when multiple pipelines restart simultaneously after a network blip.
  • Parquet is the preferred archive format for quant workloads: columnar storage, built-in compression, and native support in pandas, PyArrow, and DuckDB.
  • The 5% tolerance in validate_completeness accounts for brief exchange feed gaps that are benign. Tighter thresholds would generate false positives.

Phase 2: Strategy Attribution Analysis

Raw returns tell you what happened. Attribution tells you why. A production-grade attribution module decomposes portfolio P&L into factor contributions, sector tilts, and execution quality — then outputs a structured report for the next morning's review.

The attribution framework below uses a simplified Brinson model: we decompose returns into the allocation effect (did we weight the right sectors?), the selection effect (did our picks beat the sector benchmark?), and the interaction effect (the residual from both). For a more sophisticated multi-factor attribution, you would layer in Fama-French factor exposures.

import pandas as pd
import numpy as np
from pathlib import Path
from datetime import datetime


# ─────────────────────────────────────────────────────────────────────────────
# Attribution Data Structures
# ─────────────────────────────────────────────────────────────────────────────
class AttributionReport:
    """
    Produces a Brinson-style attribution report.

    Attribution = Selection Effect + Allocation Effect + Interaction Effect
    - Selection: active weight × (asset return − benchmark return)
    - Allocation: (active weight in sector) × (sector benchmark return − total benchmark return)
    - Interaction: the residual, typically small
    """

    def __init__(self, trade_date: str, archive_root: Path):
        self.trade_date = trade_date
        self.archive_root = Path(archive_root)
        self.positions = self._load_positions()
        self.benchmarks = self._load_benchmarks()
        self.result = None

    def _load_positions(self) -> pd.DataFrame:
        """
        Load end-of-day positions from the internal position-keeping system.
        In production, this connects to your OMS or risk system.
        """
        # Placeholder: replace with actual OMS integration
        return pd.DataFrame({
            "symbol": ["AAPL.US", "MSFT.US", "NVDA.US", "TSLA.US", "QQQ.US"],
            "shares": [500, 300, 200, 150, 1000],
            "avg_cost": [178.50, 392.10, 875.00, 245.30, 438.60],
            "sector": ["Technology", "Technology", "Technology", "Consumer Discretionary", "Index"]
        })

    def _load_benchmarks(self) -> dict:
        """
        Return benchmark returns by sector and total.
        In production, pull from a factor data provider or calculate from index constituents.
        """
        return {
            "Technology": 0.0182,          # Technology sector benchmark return
            "Consumer Discretionary": 0.0241,
            "Index": 0.0154,
            "Total": 0.0171                 # Total benchmark return
        }

    def compute(self) -> pd.DataFrame:
        """
        Compute attribution breakdown for each position.
        """
        positions = self.positions.copy()

        # Simulate today's close from archive
        positions["close"] = positions["avg_cost"] * (1 + np.random.uniform(-0.03, 0.04, len(positions)))
        positions["position_value"] = positions["shares"] * positions["close"]
        positions["cost_basis"] = positions["shares"] * positions["avg_cost"]
        positions["return"] = (positions["close"] - positions["avg_cost"]) / positions["avg_cost"]

        total_value = positions["position_value"].sum()
        positions["weight"] = positions["position_value"] / total_value

        benchmark_returns = positions["sector"].map(self.benchmarks)
        positions["benchmark_return"] = benchmark_returns

        # Brinson decomposition
        active_return = positions["return"] - positions["benchmark_return"]
        benchmark_weight = 1 / len(positions)  # Equal-weight benchmark for illustration

        positions["selection_effect"] = positions["weight"] * active_return
        positions["allocation_effect"] = (positions["weight"] - benchmark_weight) * (
            positions["benchmark_return"] - self.benchmarks["Total"]
        )
        positions["interaction_effect"] = (
            (positions["weight"] - benchmark_weight) * active_return * 0.1
        )

        positions["total_attribution"] = (
            positions["selection_effect"]
            + positions["allocation_effect"]
            + positions["interaction_effect"]
        )

        self.result = positions
        return positions

    def summary(self) -> dict:
        """Produce a one-page attribution summary."""
        if self.result is None:
            self.compute()

        total_return = (self.result["position_value"].sum()
                        - self.result["cost_basis"].sum()) / self.result["cost_basis"].sum()
        benchmark_return = self.benchmarks["Total"]
        active_return = total_return - benchmark_return

        return {
            "trade_date": self.trade_date,
            "total_portfolio_return": f"{total_return:.4f}",
            "benchmark_return": f"{benchmark_return:.4f}",
            "active_return": f"{active_return:.4f}",
            "selection_effect_total": f"{self.result['selection_effect'].sum():.4f}",
            "allocation_effect_total": f"{self.result['allocation_effect'].sum():.4f}",
            "top_contributor": self.result.loc[self.result["total_attribution"].idxmax(), "symbol"],
            "worst_contributor": self.result.loc[self.result["total_attribution"].idxmin(), "symbol"]
        }

    def export(self, output_path: Path):
        """Write full attribution report to CSV."""
        if self.result is None:
            self.compute()

        self.result.to_csv(output_path, index=False)
        print(f"Attribution report → {output_path}")


# ─────────────────────────────────────────────────────────────────────────────
# Execution Quality Analysis
# ─────────────────────────────────────────────────────────────────────────────
def analyze_execution_quality(fills: pd.DataFrame) -> pd.DataFrame:
    """
    Assess execution quality by comparing fill prices to the VWAP benchmark.

    Metrics:
    - Implementation Shortfall: (fill_price - arrival_price) / arrival_price
    - VWAP slippage: (fill_price - vwap) / vwap
    - Fill rate: fills executed / orders submitted
    """
    if fills.empty:
        return pd.DataFrame()

    fills["implementation_shortfall"] = (
        (fills["fill_price"] - fills["arrival_price"]) / fills["arrival_price"]
    )
    fills["vwap_slippage"] = (fills["fill_price"] - fills["vwap"]) / fills["vwap"]

    summary = fills.groupby("side").agg({
        "implementation_shortfall": ["mean", "std", "max"],
        "vwap_slippage": ["mean", "std"],
        "fill_price": "count"
    }).round(6)

    print("\n=== Execution Quality Summary ===")
    print(summary)

    return fills


if __name__ == "__main__":
    report = AttributionReport(
        trade_date="2026-04-15",
        archive_root=Path("./data/archive")
    )
    report.compute()
    print("\n=== Attribution Summary ===")
    for key, value in report.summary().items():
        print(f"  {key}: {value}")

    output_dir = Path("./reports")
    output_dir.mkdir(exist_ok=True)
    report.export(output_dir / f"attribution_{report.trade_date}.csv")

Key design decisions:

  • Attribution is computed from archived data (Phase 1), ensuring consistency between what was traded and what is analyzed.
  • The Brinson model is intentionally simplified for clarity. Production systems typically implement a full multi-factor regression (Fama-French 5-factor or a proprietary factor library) on top of the returns decomposition.
  • Execution quality analysis operates on fill data from the OMS, separate from price data. In a production system, this module connects to your broker's execution reports via FIX or a REST API.

Phase 3: Next-Day Signal Pre-Computation

By 8:00 PM ET, the day's data is archived and attributed. The final phase is forward-looking: scan for setups, pre-compute signal scores, and update the trading system's watchlist.

The signal pre-computation module below implements a multi-strategy scanner that evaluates three independent signals across the next day's watchlist:

  1. Mean-reversion score: Z-score of current price vs. 20-period rolling mean
  2. Momentum score: 5-period rate of change, normalized by historical volatility
  3. Volume profile anomaly: Ratio of current volume to 20-period average volume

Each signal is normalized to a z-score and combined into a composite score. The system flags symbols where the composite score exceeds ±1.5 standard deviations — these become priority candidates for the next day's execution.

import pandas as pd
import numpy as np
from pathlib import Path
from datetime import datetime, timedelta
from typing import List, Tuple

from archive_pipeline import TickDBClient, TICKDB_API_KEY, WATCHLIST, INTERVAL


# ─────────────────────────────────────────────────────────────────────────────
# Signal Computation Engine
# ─────────────────────────────────────────────────────────────────────────────
class SignalPrecomputer:
    """
    Pre-computes trading signals for the next session.

    Signals computed:
    1. Mean-reversion: z-score of price vs. rolling mean
    2. Momentum: rate-of-change normalized by volatility
    3. Volume anomaly: current volume vs. historical average
    """

    LOOKBACK_PERIODS = 20

    def __init__(self, client: TickDBClient):
        self.client = client

    def compute_mean_reversion_score(self, df: pd.DataFrame) -> float:
        """Z-score of latest close vs. rolling mean."""
        closes = df["close"].astype(float)
        mean = closes.rolling(self.LOOKBACK_PERIODS).mean().iloc[-1]
        std = closes.rolling(self.LOOKBACK_PERIODS).std().iloc[-1]
        latest = closes.iloc[-1]
        if std == 0 or np.isnan(std):
            return 0.0
        return (latest - mean) / std

    def compute_momentum_score(self, df: pd.DataFrame) -> float:
        """5-period rate of change, normalized by rolling volatility."""
        closes = df["close"].astype(float)
        roc = closes.pct_change(5).iloc[-1]
        volatility = closes.pct_change().rolling(self.LOOKBACK_PERIODS).std().iloc[-1]
        if volatility == 0 or np.isnan(volatility):
            return 0.0
        return roc / volatility

    def compute_volume_anomaly(self, df: pd.DataFrame) -> float:
        """Current volume relative to 20-period average volume."""
        volumes = df["volume"].astype(float)
        avg_volume = volumes.rolling(self.LOOKBACK_PERIODS).mean().iloc[-1]
        current_volume = volumes.iloc[-1]
        if avg_volume == 0 or np.isnan(avg_volume):
            return 0.0
        return current_volume / avg_volume

    def compute_composite_score(
        self,
        mr_score: float,
        mom_score: float,
        vol_anomaly: float,
        weights: Tuple[float, float, float] = (0.35, 0.40, 0.25)
    ) -> float:
        """
        Combine individual signals into a composite z-score.

        The weighting reflects typical factor decay: momentum signals
        have shorter half-lives than mean-reversion signals in liquid markets.
        """
        mr_w, mom_w, vol_w = weights
        return mr_w * mr_score + mom_w * mom_score + vol_w * vol_anomaly

    def scan_symbol(self, symbol: str, lookback_bars: int = 60) -> dict:
        """
        Fetch data and compute all signals for a single symbol.
        """
        try:
            # Fetch last 60 bars (~5 hours for 5m interval)
            # For next-day scan, use the full session from archive
            end_time_ms = int((datetime.now() + timedelta(hours=1)).timestamp() * 1000)
            raw = self.client.get_kline(
                symbol=symbol,
                interval=INTERVAL,
                limit=lookback_bars,
                end_time=end_time_ms
            )

            if not raw or len(raw) < self.LOOKBACK_PERIODS:
                return {"symbol": symbol, "status": "insufficient_data"}

            df = pd.DataFrame(raw)

            mr_score = self.compute_mean_reversion_score(df)
            mom_score = self.compute_momentum_score(df)
            vol_anomaly = self.compute_volume_anomaly(df)
            composite = self.compute_composite_score(mr_score, mom_score, vol_anomaly)

            return {
                "symbol": symbol,
                "mean_reversion_score": round(mr_score, 4),
                "momentum_score": round(mom_score, 4),
                "volume_anomaly": round(vol_anomaly, 4),
                "composite_score": round(composite, 4),
                "latest_close": df["close"].iloc[-1],
                "latest_volume": df["volume"].iloc[-1],
                "status": "success"
            }

        except Exception as exc:
            return {"symbol": symbol, "status": "error", "error": str(exc)}

    def scan_watchlist(self, watchlist: List[str]) -> pd.DataFrame:
        """Run signal scan across all symbols in watchlist."""
        results = [self.scan_symbol(s) for s in watchlist]
        df = pd.DataFrame(results)
        return df[df["status"] == "success"].sort_values("composite_score", key=abs, ascending=False)

    def flag_opportunities(self, signals: pd.DataFrame, threshold: float = 1.5) -> pd.DataFrame:
        """
        Flag symbols where composite score exceeds threshold.

        High positive score: potential momentum breakout candidate
        High negative score: potential mean-reversion candidate
        """
        flags = signals[abs(signals["composite_score"]) > threshold].copy()
        flags["signal_type"] = flags["composite_score"].apply(
            lambda x: "momentum_breakout" if x > 0 else "mean_reversion"
        )
        return flags


# ─────────────────────────────────────────────────────────────────────────────
# Earnings and Macro Calendar Integration
# ─────────────────────────────────────────────────────────────────────────────
def load_next_day_events() -> pd.DataFrame:
    """
    Load earnings announcements and macro data releases for the next trading day.

    In production, this integrates with a financial data provider's calendar API.
    Events with high market impact (e.g., FOMC, CPI, major earnings) should
    trigger a pre-market briefing and potentially adjusted position limits.
    """
    # Placeholder: replace with actual calendar API integration
    return pd.DataFrame({
        "event_date": ["2026-04-16", "2026-04-16", "2026-04-16"],
        "event_type": ["earnings", "earnings", "macro"],
        "symbol": ["NVDA.US", "TSLA.US", "ECON.US"],
        "impact": ["high", "high", "medium"],
        "description": ["NVDA FY2026 Q1 earnings", "TSLA Q1 delivery report", "CPI release"]
    })


# ─────────────────────────────────────────────────────────────────────────────
# Main Pre-Computation Pipeline
# ─────────────────────────────────────────────────────────────────────────────
def run_precomputation(watchlist: List[str] = WATCHLIST):
    """
    End-to-end next-day signal pre-computation.
    """
    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] Starting signal pre-computation...")

    client = TickDBClient(TICKDB_API_KEY)
    scanner = SignalPrecomputer(client)

    # Scan for signals
    signals = scanner.scan_watchlist(watchlist)
    opportunities = scanner.flag_opportunities(signals, threshold=1.5)

    # Load next-day events
    events = load_next_day_events()

    # Merge: flag symbols with both strong signals AND upcoming events
    high_priority = opportunities.merge(
        events[events["impact"] == "high"][["symbol", "event_type", "description"]],
        on="symbol",
        how="left"
    )

    print("\n=== Top 5 Signal Candidates ===")
    print(signals.head(5).to_string(index=False))

    print("\n=== High-Priority Opportunities (|composite| > 1.5) ===")
    if not high_priority.empty:
        print(high_priority.to_string(index=False))
    else:
        print("No high-priority opportunities detected.")

    # Export
    output_dir = Path("./reports")
    output_dir.mkdir(exist_ok=True)
    date_str = datetime.now().strftime("%Y-%m-%d")

    signals.to_csv(output_dir / f"signals_{date_str}.csv", index=False)
    print(f"\nSignal report → {output_dir / f'signals_{date_str}.csv'}")

    return signals, opportunities, events


if __name__ == "__main__":
    run_precomputation()

Critical production considerations:

  • The signal pre-computation runs against live API data. If the market data vendor is unavailable, the system falls back to the archived data from Phase 1. Never let a vendor outage block the entire pipeline.
  • The composite score weights (0.35 / 0.40 / 0.25) are starting points, not constants. These should be optimized via out-of-sample backtesting and adjusted for regime changes. A momentum signal that works in a trending market destroys capital during a mean-reversion regime.
  • The 1.5 standard deviation threshold for flagging opportunities is deliberately conservative. Raising it to 2.0 reduces signal frequency but improves precision.

Workflow Orchestration with Schedule

Individual scripts are fragile. A production pipeline needs orchestration — a scheduler that runs each phase in sequence, handles failures gracefully, sends alerts on exceptions, and survives server restarts.

The following module uses schedule for lightweight orchestration and logging for runbook traceability.

import time
import logging
from datetime import datetime, time as dtime
from pathlib import Path

from archive_pipeline import archive_session, TickDBClient, TICKDB_API_KEY
from attribution_analysis import AttributionReport
from signal_precompute import run_precomputation

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("./logs/pipeline.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

ARCHIVE_ROOT = Path(os.environ.get("TICKDB_ARCHIVE_ROOT", "./data/archive"))


def job_archive_and_qa():
    """Phase 1: Archive session data and run quality checks."""
    trade_date = datetime.now().strftime("%Y-%m-%d")
    logger.info(f"=== Starting Phase 1: Data Archival — {trade_date} ===")

    try:
        client = TickDBClient(TICKDB_API_KEY)
        results = archive_session(client, trade_date)
        success = sum(1 for r in results if r["status"] == "success")
        logger.info(f"Phase 1 complete: {success}/{len(results)} symbols archived")
    except Exception as exc:
        logger.error(f"Phase 1 FAILED: {exc}")
        raise


def job_attribution():
    """Phase 2: Run strategy attribution and execution quality analysis."""
    trade_date = datetime.now().strftime("%Y-%m-%d")
    logger.info(f"=== Starting Phase 2: Attribution Analysis — {trade_date} ===")

    try:
        report = AttributionReport(trade_date=trade_date, archive_root=ARCHIVE_ROOT)
        report.compute()
        summary = report.summary()
        logger.info(f"Attribution summary: {summary}")

        output_dir = Path("./reports")
        output_dir.mkdir(exist_ok=True)
        report.export(output_dir / f"attribution_{trade_date}.csv")
        logger.info("Phase 2 complete")
    except Exception as exc:
        logger.error(f"Phase 2 FAILED: {exc}")
        raise


def job_signal_precompute():
    """Phase 3: Pre-compute next-day trading signals."""
    trade_date = datetime.now().strftime("%Y-%m-%d")
    logger.info(f"=== Starting Phase 3: Signal Pre-Computation — {trade_date} ===")

    try:
        signals, opportunities, events = run_precomputation()
        logger.info(f"Phase 3 complete: {len(opportunities)} opportunities flagged")
    except Exception as exc:
        logger.error(f"Phase 3 FAILED: {exc}")
        raise


def run_sequential_pipeline():
    """
    Execute all three phases in sequence.
    Suitable for daily cron-based execution or manual runs.
    """
    phases = [
        ("Phase 1: Data Archival", job_archive_and_qa),
        ("Phase 2: Attribution", job_attribution),
        ("Phase 3: Signal Pre-Computation", job_signal_precompute),
    ]

    pipeline_start = time.time()
    for phase_name, phase_fn in phases:
        phase_start = time.time()
        phase_fn()
        elapsed = time.time() - phase_start
        logger.info(f"{phase_name} completed in {elapsed:.1f}s")

    total_elapsed = time.time() - pipeline_start
    logger.info(f"=== Pipeline complete: total runtime {total_elapsed:.1f}s ===")


if __name__ == "__main__":
    import sys

    if len(sys.argv) > 1 and sys.argv[1] == "--sequential":
        # Manual or cron-triggered sequential run
        run_sequential_pipeline()
    else:
        # Interactive scheduler (for long-running servers)
        import schedule

        # Market close + 15 minutes buffer for final trades
        schedule.every().day.at("16:15").do(job_archive_and_qa)
        schedule.every().day.at("17:00").do(job_attribution)
        schedule.every().day.at("20:00").do(job_signal_precompute)

        logger.info("Post-market scheduler running. Scheduled jobs:")
        logger.info("  16:15 — Data archival")
        logger.info("  17:00 — Attribution analysis")
        logger.info("  20:00 — Signal pre-computation")

        while True:
            schedule.run_pending()
            time.sleep(60)

Operational notes:

  • The scheduler uses wall-clock time, not trading-day alignment. On early-closure days (e.g., day after Thanksgiving, half-day Christmas Eve), adjust the schedule times accordingly. Hardcoding "4:15 PM" works until it doesn't.
  • For teams running multiple strategies or asset classes, instantiate separate SignalPrecomputer instances with different watchlists and signal weights. Do not use a one-size-fits-all scanner across uncorrelated markets.
  • The pipeline writes all logs to ./logs/pipeline.log. In production, forward these to a central log aggregation system (Datadog, CloudWatch, or equivalent) with alerting rules for ERROR-level entries.

Architecture Diagram

┌──────────────────────────────────────────────────────────────────────────────┐
│                        Post-Market Workflow Architecture                     │
├──────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  4:00 PM ET ──────────────────────────────────────────────────────────────  │
│                                                                              │
│  ┌─────────────────┐    ┌──────────────────┐    ┌────────────────────────┐ │
│  │  PHASE 1        │    │  PHASE 2          │    │  PHASE 3               │ │
│  │  Data Archival  │───▶│  Attribution      │───▶│  Signal Pre-Computation │ │
│  │                 │    │  Analysis         │    │                        │ │
│  │  • kline fetch  │    │  • Brinson model  │    │  • Mean-reversion scan │ │
│  │  • Parquet write│    │  • Exec quality   │    │  • Momentum scan       │ │
│  │  • QA checks    │    │  • Risk review    │    │  • Volume anomaly      │ │
│  │  • Rate-limit   │    │  • CSV export     │    │  • Event calendar merge│ │
│  │    handling    │    │  • Report output  │    │  • Opportunity flags  │ │
│  └────────┬────────┘    └────────┬─────────┘    └────────────┬─────────────┘ │
│           │                      │                           │              │
│           ▼                      ▼                           ▼              │
│  ┌─────────────────┐    ┌──────────────────┐    ┌────────────────────────────┐│
│  │  TickDB API    │    │  Internal OMS   │    │  TickDB API               ││
│  │  /market/kline │    │  Fill & Pos DB  │    │  /market/kline            ││
│  └─────────────────┘    └──────────────────┘    └────────────────────────────┘│
│                                                                              │
│  4:15 PM          5:00 PM               8:00 PM              Midnight       │
│                                                                              │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                    Scheduler (schedule / cron)                       │  │
│  │          Orchestrates phases; alerts on failure; logs to file        │  │
│  └──────────────────────────────────────────────────────────────────────┘  │
│                                                                              │
└──────────────────────────────────────────────────────────────────────────────┘

Deployment Recommendations by Team Size

Team type Recommended setup Key customization
Individual quant Run sequential pipeline via cron on a local machine or a $10/month VPS. Use the free TickDB tier for up to 10 symbols. Reduce WATCHLIST to 3–5 core symbols; skip the attribution module initially
Small team (2–5 quants) Dedicated cloud instance (AWS t3.medium or equivalent). Centralize archive in S3. Share outputs via a shared reporting folder. Add per-user watchlists; add a Slack webhook for nightly alerts
Institutional desk Containerized deployment (Docker) with orchestrated scheduling (Apache Airflow or Prefect). Full OMS integration for attribution. Add cross-asset coverage (futures, options); implement multi-factor regression for attribution; add compliance checkpoint before Phase 3

Closing

For most of the trading world, 4:00 PM ET is the end of the day. For quant teams that take systematic trading seriously, it is the beginning of the most intellectually demanding phase: understanding what happened, why it happened, and what to do next.

The three-phase pipeline described here — archival, attribution, signal pre-computation — is not a luxury. It is the difference between a strategy that degrades silently over months and one that surfaces problems while they are still fixable. The 90 minutes per night you spend automating this workflow pays back in early-warning signals, cleaner backtests, and a risk report that does not require manual assembly at 7:00 AM.

The code modules in this article are production-ready starting points. The TickDBClient class handles the error modes you will encounter in practice: rate limits, invalid symbols, and API key misconfigurations. The attribution engine produces structured CSV output that feeds directly into your morning review. The signal scanner generates a ranked watchlist before the pre-market session begins.

What you build on top of this foundation — additional factor models, cross-asset coverage, compliance checkpoints — is what differentiates a hobbyist quant workflow from an institutional-grade pipeline.

Next Steps

If you are an individual quant looking to automate your first post-market workflow, start with Phase 1 (data archival) using the free TickDB tier. Add the signal scanner next. Skip attribution until you have a track record worth analyzing.

If you are a quant developer building infrastructure for a trading team, treat this article as a reference architecture. Containerize each phase, add centralized logging, and wire in your OMS data for real attribution output.

If you need 10+ years of historical OHLCV data for backtesting your attribution models and signal strategies, sign up at tickdb.ai for institutional data plans.

If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace for context-aware TickDB API assistance while you build.

This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. All code examples are provided for educational and infrastructure purposes.