The closing bell rings at 4:00 PM ET. Most retail traders close their platforms and walk away. Professional quantitative teams do not.
For a systematic trading desk, the eight hours of market operation generate a flood of raw material: tick-by-tick order flow, incomplete candles waiting to close, position P&L snapshots mid-drift, factor exposures shifting in real time. None of this data is useful in its raw form. It must be captured, cleaned, aligned, and transformed into intelligence. That intelligence feeds tomorrow's decisions.
The hours between 4:00 PM and midnight are not idle. They are where edge compounds. This article dissects the post-market automation pipeline that separates a sustainable quant operation from a discretionary trader with Python scripts. We will walk through three core modules: data archival and ETL, strategy attribution analysis, and tomorrow's signal pre-computation. Production-grade code examples accompany each stage.
Why Post-Market Automation Cannot Be Manual
Before diving into implementation, it is worth addressing the organizational question: why does this workflow need automation at all?
A retail trader running one strategy on five symbols can afford to spend twenty minutes each evening reviewing their trades manually. A quant team running forty strategies across six hundred symbols cannot. Manual review at that scale introduces two compounding failure modes.
Fatigue-driven errors. Attribution analysis requires comparing realized performance against predicted signals across hundreds of positions. A human analyst making this comparison at 10:00 PM after a volatile session will miss correlations that a script will detect on the third pass.
Opportunity cost. Pre-computing tomorrow's signal candidates is not optional if your strategy has a lookback window requirement or depends on overnight earnings data. A team that computes these signals manually every morning is arriving at the starting line thirty minutes behind a team that wakes up to pre-computed candidates in a dashboard.
The operational discipline to run this pipeline every single day—even on low-volatility Fridays when "nothing happened"—is what separates systematic returns from systematic drift.
The Post-Market Pipeline: Three Stages
The end-of-day automation pipeline consists of three logically distinct stages, each with its own dependencies and failure modes:
| Stage | Input | Output | Typical runtime |
|---|---|---|---|
| Data archival & ETL | Raw market data (kline, trades, depth) | Cleaned, time-aligned datasets | 15–45 min |
| Strategy attribution | Cleaned datasets + signal logs | Factor P&L decomposition, anomaly reports | 10–30 min |
| Signal pre-computation | Cleaned datasets + overnight events | Tomorrow's candidate signal list | 20–60 min |
These stages have a linear dependency chain. Stage 2 cannot run until Stage 1 completes. Stage 3 waits on Stage 1's cleaned datasets. This sequential nature makes the pipeline an ideal candidate for task orchestration rather than ad-hoc shell scripts.
Stage 1: Data Archival and ETL Pipeline
The Data Quality Problem
Market data arrives from multiple venues with timestamp jitter, misaligned session boundaries, and occasional gaps. A US equity kline from TickDB covers the official trading session (9:30–16:00 ET), but your strategy may have traded during pre-market (4:00–9:30 ET) or after-hours (16:00–20:00 ET). The ETL layer must:
- Fetch all available session data for the trading day
- Align timestamps to a single reference timezone (UTC is standard)
- Detect and flag anomalous candles (e.g., a candle spanning 4:00–9:30 with a gap at 9:30)
- Persist the cleaned dataset with a partition key (date, symbol) for downstream queries
ETL Implementation
The following production-grade script implements the archival and cleaning layer. It uses a lightweight scheduling mechanism and includes retry logic with exponential backoff for API resilience.
"""
Post-market ETL pipeline for TickDB market data archival.
Fetches daily OHLCV data, cleans timestamps, and persists to local storage.
"""
import os
import json
import time
import logging
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Optional
import requests
# ─── Configuration ────────────────────────────────────────────────────────────
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
if not TICKDB_API_KEY:
raise ValueError("TICKDB_API_KEY environment variable is not set")
BASE_URL = "https://api.tickdb.ai/v1"
WATCHLIST = [
"AAPL.US", "MSFT.US", "NVDA.US", "TSLA.US", "SPY.US",
"QQQ.US", "IWM.US", "GLD.US", "TLT.US", "BTC/USD"
]
# ─── Logging setup ─────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
handlers=[
logging.StreamHandler(),
logging.FileHandler("/var/log/tickdb_etl.log")
]
)
logger = logging.getLogger("etl_pipeline")
# ─── API client with production-grade resilience ───────────────────────────────
def fetch_with_retry(
url: str,
params: Optional[dict] = None,
max_retries: int = 5,
timeout: tuple = (3.05, 27)
) -> dict:
"""
Fetch data from TickDB with exponential backoff and jitter.
Handles rate-limit responses (code 3001) with Retry-After respect.
"""
headers = {"X-API-Key": TICKDB_API_KEY}
attempt = 0
while attempt < max_retries:
try:
response = requests.get(url, headers=headers, params=params, timeout=timeout)
data = response.json()
# Check for rate limiting
code = data.get("code", 0)
if code == 3001:
retry_after = int(response.headers.get("Retry-After", 5))
logger.warning(
f"Rate limited. Waiting {retry_after}s before retry "
f"(attempt {attempt + 1}/{max_retries})"
)
time.sleep(retry_after)
attempt += 1
continue
# Check for successful response
if code == 0:
return data.get("data", {})
# Handle known error codes
error_messages = {
1001: "Invalid API key",
1002: "Missing API key",
2002: "Symbol not found",
}
if code in error_messages:
raise ValueError(f"TickDB error {code}: {error_messages[code]}")
raise RuntimeError(f"TickDB error {code}: {data.get('message', 'Unknown error')}")
except requests.exceptions.Timeout:
logger.warning(
f"Request timeout (attempt {attempt + 1}/{max_retries}). Retrying..."
)
attempt += 1
time.sleep(2 ** attempt + random.uniform(0, 1)) # Exponential backoff + jitter
except requests.exceptions.RequestException as e:
logger.error(f"Request failed: {e}")
raise
raise RuntimeError(f"Failed after {max_retries} retries")
def get_available_symbols() -> list:
"""Fetch list of available symbols from TickDB."""
url = f"{BASE_URL}/symbols/available"
data = fetch_with_retry(url)
return [s.get("symbol") for s in data.get("symbols", []) if s.get("symbol")]
def fetch_daily_kline(symbol: str, target_date: datetime) -> dict:
"""
Fetch 1-minute kline data for a specific symbol and date.
Note: TickDB kline endpoint returns data for completed periods.
For intraday pre-market and after-hours, use extended session parameters.
"""
url = f"{BASE_URL}/market/kline"
# Calculate the date range: start of previous day to end of target day
# This ensures we capture any extended-hours activity
start_time = int((target_date - timedelta(days=1)).replace(
hour=4, minute=0, second=0, tzinfo=timezone.utc
).timestamp())
end_time = int(target_date.replace(
hour=20, minute=0, second=0, tzinfo=timezone.utc
).timestamp())
params = {
"symbol": symbol,
"interval": "1m",
"start": start_time,
"end": end_time,
"limit": 1500 # Max records per request
}
return fetch_with_retry(url, params=params)
def clean_kline_data(raw_data: dict, symbol: str, target_date: datetime) -> list:
"""
Clean and validate kline data.
- Converts timestamps to UTC-aware datetime objects
- Filters to the target trading date
- Detects gaps and flags anomalous candles
"""
klines = raw_data.get("klines", [])
if not klines:
logger.warning(f"No kline data returned for {symbol}")
return []
cleaned = []
target_date_str = target_date.date().isoformat()
previous_close = None
for k in klines:
ts = datetime.fromtimestamp(k["t"] / 1000, tz=timezone.utc)
open_price = float(k["o"])
high_price = float(k["h"])
low_price = float(k["l"])
close_price = float(k["c"])
volume = int(k["v"])
# Skip candles not belonging to the target date
if ts.date().isoformat() != target_date_str:
continue
# Anomaly detection: check for obviously wrong data
anomaly_flags = []
if high_price < low_price:
anomaly_flags.append("HIGH_BELOW_LOW")
if high_price < open_price or high_price < close_price:
anomaly_flags.append("HIGH_BELOW_OPEN_OR_CLOSE")
if low_price > open_price or low_price > close_price:
anomaly_flags.append("LOW_ABOVE_OPEN_OR_CLOSE")
# Check for gap from previous candle
if previous_close is not None:
gap_pct = abs(close_price - previous_close) / previous_close
if gap_pct > 0.05: # Flag >5% gap
anomaly_flags.append(f"LARGE_GAP_{gap_pct:.2%}")
cleaned.append({
"symbol": symbol,
"timestamp": ts.isoformat(),
"open": open_price,
"high": high_price,
"low": low_price,
"close": close_price,
"volume": volume,
"anomalies": "|".join(anomaly_flags) if anomaly_flags else None
})
previous_close = close_price
logger.info(f"Cleaned {len(cleaned)} candles for {symbol} on {target_date_str}")
return cleaned
def persist_to_sqlite(cleaned_data: list, db_path: str = "/data/tickdb_archive.db"):
"""Persist cleaned kline data to SQLite with date-symbol partitioning."""
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Create table if not exists
cursor.execute("""
CREATE TABLE IF NOT EXISTS kline_daily (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
timestamp TEXT NOT NULL,
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume INTEGER NOT NULL,
anomalies TEXT,
loaded_at TEXT NOT NULL,
UNIQUE(symbol, timestamp)
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_symbol_timestamp
ON kline_daily(symbol, timestamp)
""")
# Insert data with conflict handling (upsert)
loaded_at = datetime.now(timezone.utc).isoformat()
inserted = 0
for record in cleaned_data:
try:
cursor.execute("""
INSERT OR REPLACE INTO kline_daily
(symbol, timestamp, open, high, low, close, volume, anomalies, loaded_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
record["symbol"], record["timestamp"], record["open"], record["high"],
record["low"], record["close"], record["volume"], record["anomalies"],
loaded_at
))
inserted += 1
except sqlite3.Error as e:
logger.error(f"Database insert failed for {record['symbol']}: {e}")
conn.commit()
conn.close()
logger.info(f"Persisted {inserted}/{len(cleaned_data)} records to {db_path}")
return inserted
# ⚠️ For production HFT workloads, consider PostgreSQL with TimescaleDB extension
# or a dedicated time-series database like InfluxDB for better query performance.
def run_etl_pipeline(trading_date: Optional[datetime] = None):
"""
Main ETL orchestration function.
Run this via a scheduler (cron, Airflow, Prefect) after market close.
"""
if trading_date is None:
# Default to previous trading day
trading_date = datetime.now(timezone.utc) - timedelta(days=1)
# Adjust for weekend: if today is Monday, last trading day was Friday
if trading_date.weekday() == 0: # Monday
trading_date -= timedelta(days=2)
logger.info(f"Starting ETL pipeline for trading date: {trading_date.date()}")
total_candles = 0
failed_symbols = []
for symbol in WATCHLIST:
try:
raw_data = fetch_daily_kline(symbol, trading_date)
cleaned = clean_kline_data(raw_data, symbol, trading_date)
persisted = persist_to_sqlite(cleaned)
total_candles += persisted
except Exception as e:
logger.error(f"ETL failed for {symbol}: {e}")
failed_symbols.append(symbol)
continue
# Rate limiting: pause between requests to avoid triggering 3001
time.sleep(0.5)
logger.info(
f"ETL pipeline complete. "
f"Total candles: {total_candles}, Failed symbols: {len(failed_symbols)}"
)
return {
"status": "success" if not failed_symbols else "partial_failure",
"total_candles": total_candles,
"failed_symbols": failed_symbols,
"trading_date": trading_date.date().isoformat()
}
if __name__ == "__main__":
result = run_etl_pipeline()
print(json.dumps(result, indent=2))
ETL Pipeline Architecture
┌─────────────────────────────────────────────────────────────────┐
│ SCHEDULER (cron / Airflow) │
│ Trigger: 0 16 * * 1-5 (4:00 PM ET weekdays) │
└───────────────────────────────┬─────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ STAGE 1: Data Archival │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ TickDB API │──▶│ Cleaning & │──▶│ SQLite / TimescaleDB │ │
│ │ /kline │ │ Validation │ │ (Partitioned Storage) │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
│ │ │ │ │
│ │ Timestamp Date-Symbol │
│ │ alignment Partitioning │
└───────┼───────────────────┼────────────────────┼────────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────┐
│ STAGE 2: Attribution │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Signal Log │ │ P&L │ │ Factor Decomposition │ │
│ │ Storage │──▶│ Attribution │──▶│ & Anomaly Report │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│ │
▼ ▼
┌─────────────────────────────────────────────────────────────────┐
│ STAGE 3: Signal Pre-Computation │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Earnings │ │ Feature │ │ Tomorrow's │ │
│ │ Calendar API │──▶│ Engineering │──▶│ Candidate List │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
Stage 2: Strategy Attribution Analysis
Attribution answers the question: "Where did the P&L come from?" Without systematic attribution, a quant team cannot distinguish skill from luck, cannot identify factor decay, and cannot diagnose why a strategy that worked in Q1 blew up in Q3.
What Good Attribution Looks Like
A production attribution report answers three questions:
What factors drove returns? For a US equity long-short strategy, the canonical decomposition is: market beta, sector exposure, size factor, value factor, momentum, and idiosyncratic alpha.
Which signals fired as predicted? Your strategy generated N signals today. How many of those signals produced the expected directional move? What was the average magnitude of the predicted vs. realized move?
Where did the strategy bleed? Transaction costs, slippage, and spread costs are expected. But anomalous losses—positions that moved 3σ against you in a low-volatility environment—warrant investigation.
Attribution Analysis Implementation
"""
Strategy attribution analysis module.
Decomposes daily P&L into factor contributions and signal performance metrics.
"""
import sqlite3
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import pandas as pd
import numpy as np
logger = logging.getLogger("attribution")
def load_signal_log(db_path: str, trading_date: str) -> pd.DataFrame:
"""
Load today's signal log from the trading database.
Expected schema: symbol, timestamp, signal_type, signal_value, predicted_direction
"""
conn = sqlite3.connect(db_path)
query = """
SELECT
symbol,
timestamp,
signal_type,
signal_value,
predicted_direction,
position_size,
entry_price,
exit_price,
realized_pnl
FROM signal_log
WHERE DATE(timestamp) = ?
ORDER BY timestamp
"""
df = pd.read_sql_query(query, conn, params=(trading_date,))
conn.close()
if df.empty:
logger.warning(f"No signal log entries found for {trading_date}")
return df
logger.info(f"Loaded {len(df)} signal entries for {trading_date}")
return df
def load_market_returns(db_path: str, trading_date: str) -> pd.DataFrame:
"""
Load benchmark returns for the trading date.
For US equities, use SPY as the market proxy.
"""
conn = sqlite3.connect(db_path)
query = """
SELECT
symbol,
timestamp,
open,
close,
(close - open) / open AS intraday_return
FROM kline_daily
WHERE symbol IN ('SPY.US', 'QQQ.US', 'IWM.US')
AND DATE(timestamp) = ?
"""
df = pd.read_sql_query(query, conn, params=(trading_date,))
conn.close()
return df.pivot(index="timestamp", columns="symbol", values="intraday_return")
def compute_factor_attribution(
signals: pd.DataFrame,
market_returns: pd.DataFrame,
benchmark_symbol: str = "SPY.US"
) -> dict:
"""
Decompose strategy returns into factor contributions.
Simplified model: strategy_return ≈ β_market + β_sector + α_idiosyncratic
"""
if signals.empty or market_returns.empty:
return {"error": "Insufficient data for attribution"}
# Merge signals with market returns
signals["timestamp"] = pd.to_datetime(signals["timestamp"])
market_returns.index = pd.to_datetime(market_returns.index)
merged = signals.merge(
market_returns[benchmark_symbol],
left_on="timestamp",
right_index=True,
how="left",
suffixes=("", "_market")
).rename(columns={benchmark_symbol: "market_return"})
# Compute realized return from position
merged["realized_return"] = np.where(
merged["predicted_direction"] == "long",
(merged["exit_price"] - merged["entry_price"]) / merged["entry_price"],
(merged["entry_price"] - merged["exit_price"]) / merged["entry_price"]
)
# Factor attribution: CAPM-style decomposition
# Strategy beta estimate (simplified: 1.0 if strategy holds market exposure)
strategy_returns = merged["realized_return"].dropna()
if len(strategy_returns) < 5:
return {"error": "Insufficient signal count for attribution"}
market_returns_clean = merged.loc[strategy_returns.index, "market_return"].dropna()
common_index = strategy_returns.index.intersection(market_returns_clean.index)
if len(common_index) < 5:
return {"error": "Insufficient overlapping data points"}
strategy_aligned = strategy_returns.loc[common_index]
market_aligned = market_returns_clean.loc[common_index]
# OLS regression: strategy_return = α + β * market_return + ε
X = np.column_stack([np.ones(len(market_aligned)), market_aligned.values])
y = strategy_aligned.values
# Manual OLS (production code would use statsmodels)
XtX_inv = np.linalg.inv(X.T @ X)
beta_coefficients = XtX_inv @ X.T @ y
alpha = beta_coefficients[0]
beta = beta_coefficients[1]
total_return = strategy_returns.sum()
market_contribution = beta * market_aligned.mean() * len(strategy_returns)
alpha_contribution = alpha * len(strategy_returns)
idiosyncratic_contribution = total_return - market_contribution - alpha_contribution
return {
"total_return": float(total_return),
"market_contribution": float(market_contribution),
"alpha_contribution": float(alpha_contribution),
"idiosyncratic_contribution": float(idiosyncratic_contribution),
"estimated_beta": float(beta),
"estimated_alpha": float(alpha),
"signal_count": len(signals),
"active_signals": len(strategy_returns)
}
def compute_signal_performance(signals: pd.DataFrame) -> dict:
"""
Analyze signal prediction accuracy and magnitude.
"""
if signals.empty:
return {"error": "No signals to analyze"}
signals["correct_direction"] = np.sign(
(signals["exit_price"] - signals["entry_price"]) * (
1 if signals["predicted_direction"] == "long" else -1
)
) == 1
direction_accuracy = signals["correct_direction"].mean()
avg_win = signals.loc[signals["correct_direction"], "realized_pnl"].mean()
avg_loss = signals.loc[~signals["correct_direction"], "realized_pnl"].mean()
profit_factor = abs(avg_win / avg_loss) if avg_loss != 0 else np.inf
# Magnitude analysis: how close was the realized move to the predicted magnitude?
signals["predicted_magnitude"] = signals["signal_value"] # Example: signal_value is the predicted return
signals["realized_magnitude"] = np.where(
signals["predicted_direction"] == "long",
(signals["exit_price"] - signals["entry_price"]) / signals["entry_price"],
(signals["entry_price"] - signals["exit_price"]) / signals["entry_price"]
)
magnitude_error = (signals["realized_magnitude"] - signals["predicted_magnitude"]).abs().mean()
return {
"direction_accuracy": float(direction_accuracy),
"avg_win": float(avg_win) if not np.isnan(avg_win) else None,
"avg_loss": float(avg_loss) if not np.isnan(avg_loss) else None,
"profit_factor": float(profit_factor) if not np.isinf(profit_factor) else None,
"avg_magnitude_error": float(magnitude_error),
"total_signals": len(signals),
"winning_signals": int(signals["correct_direction"].sum()),
"losing_signals": int((~signals["correct_direction"]).sum())
}
def detect_anomalies(signals: pd.DataFrame, threshold_sigma: float = 2.5) -> list:
"""
Flag positions with losses exceeding a threshold given normal volatility.
"""
if signals.empty:
return []
losses = signals.loc[signals["realized_pnl"] < 0, "realized_pnl"]
if len(losses) < 10:
return []
# Compute rolling volatility of losses
mean_loss = losses.mean()
std_loss = losses.std()
anomaly_threshold = mean_loss - (threshold_sigma * std_loss)
anomalies = signals.loc[
signals["realized_pnl"] < anomaly_threshold,
["symbol", "timestamp", "realized_pnl", "position_size", "anomaly_reason"]
].to_dict("records")
return [
{
**a,
"severity": "high" if a["realized_pnl"] < anomaly_threshold * 1.5 else "medium",
"loss_sigma": float((a["realized_pnl"] - mean_loss) / std_loss)
}
for a in anomalies
]
def generate_attribution_report(
db_path: str,
trading_date: str,
output_path: str = "/data/reports/"
) -> dict:
"""
Main attribution orchestration: generate complete daily report.
"""
logger.info(f"Generating attribution report for {trading_date}")
signals = load_signal_log(db_path, trading_date)
market_returns = load_market_returns(db_path, trading_date)
report = {
"trading_date": trading_date,
"generated_at": datetime.now(timezone.utc).isoformat(),
"factor_attribution": compute_factor_attribution(signals, market_returns),
"signal_performance": compute_signal_performance(signals),
"anomalies": detect_anomalies(signals)
}
# Persist report
Path(output_path).mkdir(parents=True, exist_ok=True)
report_file = Path(output_path) / f"attribution_{trading_date}.json"
with open(report_file, "w") as f:
json.dump(report, f, indent=2, default=str)
logger.info(f"Attribution report saved to {report_file}")
# Print summary
print(f"\n{'='*60}")
print(f"ATTRIBUTION REPORT — {trading_date}")
print(f"{'='*60}")
fa = report["factor_attribution"]
if "error" not in fa:
print(f"\nFactor Attribution:")
print(f" Total Return: {fa['total_return']:>10.4f}")
print(f" Market Contribution:{fa['market_contribution']:>10.4f}")
print(f" Alpha: {fa['alpha_contribution']:>10.4f}")
print(f" Idiosyncratic: {fa['idiosyncratic_contribution']:>10.4f}")
sp = report["signal_performance"]
if "error" not in sp:
print(f"\nSignal Performance:")
print(f" Direction Accuracy: {sp['direction_accuracy']:>10.1%}")
print(f" Total Signals: {sp['total_signals']:>10}")
print(f" Win Rate: {sp['winning_signals'] / max(sp['total_signals'], 1):>10.1%}")
if report["anomalies"]:
print(f"\nAnomalies Detected: {len(report['anomalies'])}")
for a in report["anomalies"][:5]:
print(f" [{a['severity'].upper()}] {a['symbol']}: ${a['realized_pnl']:.2f}")
print(f"{'='*60}\n")
return report
if __name__ == "__main__":
import sys
trading_date = sys.argv[1] if len(sys.argv) > 1 else (
datetime.now(timezone.utc) - timedelta(days=1)
).date().isoformat()
result = generate_attribution_report(
db_path="/data/tickdb_archive.db",
trading_date=trading_date,
output_path="/data/reports/"
)
Stage 3: Tomorrow's Signal Pre-Computation
Pre-computation is the competitive moat that systematic traders often overlook. If your strategy requires overnight earnings data, macroeconomic announcements, or sector rotation signals computed from the prior day's close, pre-computing these before the market opens is not a convenience—it is a structural requirement.
What to Pre-Compute
| Pre-computation target | Data source | Output |
|---|---|---|
| Earnings-move candidates | Earnings calendar API | List of symbols with IV surface, implied move, and historical move distribution |
| Sector rotation signals | Prior day sector returns | Rotating sector rankings (momentum + mean reversion hybrid) |
| Gap-fill probability | Pre-market / after-hours price action | Probability of gap-fill based on historical z-score |
| Overnight funding rate impact | Interest rate calendars | Adjusted position sizing for carry strategies |
| Options expiry pinning | Historical options open interest | Symbols with high gamma exposure near current price |
Signal Pre-Computation Implementation
"""
Tomorrow's signal pre-computation module.
Fetches overnight events and computes signal candidates before market open.
"""
import os
import json
import time
import logging
from datetime import datetime, timedelta, timezone, date
from pathlib import Path
from typing import Optional
import requests
logger = logging.getLogger("signal_precompute")
# ─── Configuration ────────────────────────────────────────────────────────────
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")
EPS_REPORT_API_KEY = os.environ.get("EPS_REPORT_API_KEY") # Third-party earnings API
BASE_URL = "https://api.tickdb.ai/v1"
# ─── Earnings Calendar Integration ────────────────────────────────────────────
def fetch_earnings_calendar(target_date: date) -> list:
"""
Fetch earnings announcements for the target date.
In production, replace with your preferred earnings data provider.
"""
# Example: fetch from a third-party API
# This is a placeholder that would connect to Polygon, Alpha Vantage, etc.
url = f"https://api.example.com/v1/earnings"
params = {
"date": target_date.isoformat(),
"apiKey": EPS_REPORT_API_KEY
}
try:
response = requests.get(url, params=params, timeout=(3.05, 10))
response.raise_for_status()
return response.json().get("earnings", [])
except requests.exceptions.RequestException as e:
logger.warning(f"Failed to fetch earnings calendar: {e}")
return []
def compute_implied_move_options(
symbol: str,
current_price: float,
risk_free_rate: float = 0.0525,
days_to_expiry: int = 1
) -> dict:
"""
Estimate implied move using a simplified at-the-money straddle model.
C ≈ 0.4 * σ * S * √(T) → σ ≈ C / (0.4 * S * √(T))
where C is the ATM call/put price, S is stock price, T is time in years.
Note: For production, fetch actual ATM options prices from an options data provider.
This function provides a rough estimate when options data is unavailable.
"""
# Placeholder: assume ATM straddle costs 1.5% of stock price (rough estimate)
estimated_straddle_pct = 0.015
estimated_straddle_price = current_price * estimated_straddle_pct
time_to_expiry_years = days_to_expiry / 252
implied_volatility = estimated_straddle_price / (0.4 * current_price * (time_to_expiry_years ** 0.5))
# Approximate 1-standard-deviation move
move_distance = implied_volatility * current_price * (time_to_expiry_years ** 0.5)
move_pct = (move_distance / current_price) * 100
return {
"symbol": symbol,
"current_price": current_price,
"implied_volatility": implied_volatility,
"estimated_move_distance": move_distance,
"estimated_move_pct": move_pct,
"upper_bound": current_price + move_distance,
"lower_bound": current_price - move_distance,
"model_note": "Simplified estimate — use actual options data for production"
}
def compute_gap_fill_probability(
symbol: str,
prev_close: float,
current_price: float,
historical_closes: list,
lookback_days: int = 20
) -> dict:
"""
Estimate probability of gap-fill based on historical z-score analysis.
If the gap is larger than historical norms, gap-fill probability decreases.
"""
if len(historical_closes) < lookback_days:
return {"symbol": symbol, "error": "Insufficient historical data"}
# Compute historical daily returns
returns = [
(t - p) / p
for p, t in zip(historical_closes[:-1], historical_closes[1:])
]
mean_return = sum(returns) / len(returns)
variance = sum((r - mean_return) ** 2 for r in returns) / (len(returns) - 1)
std_return = variance ** 0.5
# Compute z-score of the current gap
gap = (current_price - prev_close) / prev_close
z_score = (gap - mean_return) / std_return if std_return > 0 else 0
# Probability estimate using normal distribution approximation
# For |z| > 2, gap-fill probability drops significantly
from math import erf
gap_fill_prob = erf(abs(z_score) / (2 ** 0.5)) if abs(z_score) > 0 else 1.0
return {
"symbol": symbol,
"prev_close": prev_close,
"current_price": current_price,
"gap_pct": gap * 100,
"z_score": z_score,
"gap_fill_probability": gap_fill_prob if gap_fill_prob <= 1 else 1.0,
"confidence": "high" if abs(z_score) < 2 else ("medium" if abs(z_score) < 3 else "low")
}
def compute_sector_momentum_signals(
db_path: str,
sector_etfs: dict,
lookback_days: int = 5
) -> list:
"""
Rank sector ETFs by momentum and mean-reversion signals.
"""
import sqlite3
conn = sqlite3.connect(db_path)
signals = []
for sector_name, ticker in sector_etfs.items():
# Fetch recent closes
query = """
SELECT close, timestamp FROM kline_daily
WHERE symbol = ?
ORDER BY timestamp DESC
LIMIT ?
"""
df = pd.read_sql_query(query, conn, params=(ticker, lookback_days + 1))
conn.close()
if len(df) < lookback_days + 1:
continue
df = df.sort_values("timestamp")
closes = df["close"].tolist()
# Momentum: 5-day return
momentum = (closes[-1] - closes[0]) / closes[0]
# Mean reversion: distance from 20-day MA (if available)
if len(closes) >= 20:
ma_20 = sum(closes[:20]) / 20
distance_from_ma = (closes[-1] - ma_20) / ma_20
else:
distance_from_ma = 0
signals.append({
"sector": sector_name,
"ticker": ticker,
"momentum_5d": momentum,
"distance_from_ma": distance_from_ma,
"signal_strength": momentum - 0.5 * distance_from_ma, # Hybrid score
"recommendation": "overweight" if momentum > 0.01 else (
"underweight" if momentum < -0.01 else "neutral"
)
})
# Rank by signal strength
signals.sort(key=lambda x: x["signal_strength"], reverse=True)
return signals
def generate_tomorrow_candidate_list(
db_path: str,
target_date: date,
output_path: str = "/data/signal_candidates/"
) -> dict:
"""
Main pre-computation orchestration: generate tomorrow's signal candidates.
Run this at 5:00 PM ET (after Stage 1 and Stage 2 complete).
"""
logger.info(f"Generating signal candidates for {target_date}")
# Step 1: Fetch overnight events
earnings = fetch_earnings_calendar(target_date)
# Step 2: Compute earnings move estimates
earnings_candidates = []
for company in earnings:
symbol = company["symbol"]
# Fetch current price from TickDB (use today's close as proxy for tomorrow's open)
try:
current_price = fetch_current_price(symbol)
except Exception as e:
logger.warning(f"Could not fetch price for {symbol}: {e}")
continue
implied_move = compute_implied_move_options(
symbol, current_price, days_to_expiry=1
)
earnings_candidates.append({
"symbol": symbol,
"announcement_time": company.get("time", "AMC"), # BMC = before market close
"implied_move_pct": implied_move["estimated_move_pct"],
"upper_bound": implied_move["upper_bound"],
"lower_bound": implied_move["lower_bound"],
"confidence": "use_actual_options_data"
})
# Step 3: Compute sector rotation signals
sector_etfs = {
"Technology": "XLK.US",
"Healthcare": "XLV.US",
"Financials": "XLF.US",
"Energy": "XLE.US",
"Consumer Discretionary": "XLY.US"
}
sector_signals = compute_sector_momentum_signals(db_path, sector_etfs)
# Step 4: Assemble candidate report
report = {
"target_date": target_date.isoformat(),
"generated_at": datetime.now(timezone.utc).isoformat(),
"earnings_candidates": earnings_candidates,
"sector_signals": sector_signals,
"summary": {
"total_earnings_symbols": len(earnings_candidates),
"top_sector": sector_signals[0]["sector"] if sector_signals else None,
"high_confidence_gaps": len([
c for c in earnings_candidates
if c.get("confidence") == "high"
])
}
}
# Persist report
Path(output_path).mkdir(parents=True, exist_ok=True)
report_file = Path(output_path) / f"signal_candidates_{target_date}.json"
with open(report_file, "w") as f:
json.dump(report, f, indent=2, default=str)
logger.info(f"Signal candidates saved to {report_file}")
return report
def fetch_current_price(symbol: str) -> float:
"""Fetch today's close price from TickDB as a proxy for tomorrow's open."""
url = f"{BASE_URL}/market/kline/latest"
params = {"symbol": symbol, "interval": "1d"}
headers = {"X-API-Key": TICKDB_API_KEY}
response = requests.get(url, headers=headers, params=params, timeout=(3.05, 10))
data = response.json()
if data.get("code") == 0:
klines = data.get("data", {}).get("klines", [])
if klines:
return float(klines[-1]["c"])
raise ValueError(f"Could not fetch current price for {symbol}")
# ─── Scheduler Integration ──────────────────────────────────────────────────────
def run_scheduler():
"""
Example: how to run all three stages via a Python scheduler.
For production, replace with APScheduler, Celery, or Airflow.
"""
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
scheduler = BlockingScheduler()
# Stage 1: ETL — run at 4:05 PM ET (5 minutes after close)
scheduler.add_job(
run_etl_pipeline,
CronTrigger(hour=16, minute=5, timezone="America/New_York"),
id="stage1_etl",
replace_existing=True
)
# Stage 2: Attribution — run at 4:30 PM ET
scheduler.add_job(
generate_attribution_report,
CronTrigger(hour=16, minute=30, timezone="America/New_York"),
id="stage2_attribution",
replace_existing=True
)
# Stage 3: Signal pre-computation — run at 5:00 PM ET
scheduler.add_job(
generate_tomorrow_candidate_list,
CronTrigger(hour=17, minute=0, timezone="America/New_York"),
id="stage3_precompute",
replace_existing=True
)
logger.info("Post-market scheduler started. Press Ctrl+C to exit.")
scheduler.start()
if __name__ == "__main__":
import pandas as pd # Ensure pandas is imported for sector signals
target = datetime.now(timezone.utc).date() + timedelta(days=1)
result = generate_tomorrow_candidate_list(
db_path="/data/tickdb_archive.db",
target_date=target,
output_path="/data/signal_candidates/"
)
print(json.dumps(result["summary"], indent=2))
Operational Considerations
Error Handling and Alerts
No pipeline is complete without observability. Every stage should emit alerts on failure. A pipeline that silently fails at 4:05 PM and leaves you with no attribution report at 8:00 PM is worse than no pipeline at all.
def send_alert(message: str, severity: str = "warning"):
"""Send alert to Slack, PagerDuty, or email."""
webhook_url = os.environ.get("ALERT_WEBHOOK_URL")
if not webhook_url:
logger.warning(f"ALERT (no webhook configured): [{severity}] {message}")
return
payload = {
"text": f"[{severity.upper()}] {message}",
"attachments": [{
"color": "danger" if severity == "critical" else "warning",
"fields": [
{"title": "Pipeline Stage", "value": "Post-Market ETL", "short": True},
{"title": "Timestamp", "value": datetime.now(timezone.utc).isoformat(), "short": True}
]
}]
}
try:
requests.post(webhook_url, json=payload, timeout=(3.05, 10))
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send alert: {e}")
Backtest Disclosure
The strategies discussed in this article—earnings gap-fill probability, sector momentum rotation, implied move estimation—are hypothetical frameworks for signal generation. They have not been backtested against historical data within this article. Any reader considering implementing these signals should conduct their own backtesting with appropriate slippage, commission, and market impact assumptions.
Closing
The closing bell is not an ending. It is the start of a second shift—one that runs on servers instead of screens, on data instead of intuition. The teams that run this shift systematically, without gaps, without shortcuts, are the ones whose strategies survive regime changes, factor decay, and the inevitable drawdowns that every systematic approach encounters.
The pipeline above is not a finished product. It is a framework. The specific signals, factor models, and data sources will vary based on your strategy, your risk tolerance, and your market focus. But the discipline of running it every single day, without exception—that is non-negotiable.
If you are building systematic strategies without a post-market automation layer, you are leaving alpha on the table every single night.
Next Steps
If you are an individual quant trader looking to systematize your workflow, start with Stage 1 (the ETL pipeline). Archive three months of kline data and run your own attribution on paper trades before adding Stage 2 to production.
If you want to run this strategy yourself, sign up at tickdb.ai to access 10+ years of cleaned, time-aligned US equity OHLCV data via the /kline endpoint. The free tier includes sufficient data for single-symbol backtesting.
If you are a quant team evaluating infrastructure, reach out to enterprise@tickdb.ai for institutional data plans that include extended-hours data, depth snapshots, and dedicated API support.
If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace for integrated market data access in your development environment.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.