QuantTools is a comprehensive Python framework for quantitative trading research, backtesting, and portfolio management with a focus on options trading strategies. Built around an event-driven architecture, it enables realistic simulation of trading workflows including signal generation, risk management, order execution, and performance analysis.
- What Problems Does This Solve?
- Features
- Installation
- Repository Structure
- Core Concepts
- Quickstart
- How-To Examples
- Extended Examples: trade/ Module Deep Dive
- 8. Option Pricing with Black-Scholes
- 9. Calculating Greeks (Delta, Gamma, Vega, Theta)
- 10. Implied Volatility Calculation
- 11. Using Decorators for Performance Monitoring
- 12. Context Manager for Time Window Management
- 13. Legacy Vectorized Backtester (PTBacktester)
- 14. Volatility Surface Modeling
- 15. Custom Technical Indicators
- 16. Thread Pool for Parallel Processing
- 17. Working with Configuration Files
- 18. P&L Attribution and Greeks Decomposition
- Samples
- Troubleshooting
- Contributing
- License
- Authors
- Acknowledgments
QuantTools addresses key challenges in algorithmic trading development:
-
Realistic Backtesting: Traditional vectorized backtests ignore market microstructure, execution timing, and position management complexities. QuantTools uses an event-driven engine that processes trades chronologically with T+N settlement, slippage modeling, and realistic order flow.
-
Options-First Design: Most backtesting frameworks focus on equities. QuantTools provides native support for options strategies with Greek-based risk management, option chain analysis, position rolling, and exercise handling.
-
Modular Risk Management: Separates risk logic from strategy logic, allowing position limits, Greek constraints, and sizing rules to be configured independently via "cogs" (pluggable risk modules).
-
Data Infrastructure: Includes caching mechanisms, market data management, and timeseries handling optimized for backtesting large option universes across multi-year periods.
-
Production-Ready Patterns: Code structure mirrors production trading systems with clear separation between data handlers, strategy signals, portfolio state, execution simulation, and risk controls.
-
Event-Driven Backtesting Engine (
EventDriven/)- Queue-based event processing (MarketEvent, SignalEvent, OrderEvent, FillEvent)
- Realistic T+N settlement delays and slippage modeling
- Support for corporate actions (dividends, splits, assignments)
- Position-level P&L tracking and attribution
-
Portfolio Management (
EventDriven/new_portfolio.py)- Real-time position tracking for options and equities
- Holdings valuation with mark-to-market updates
- Trade ledger with complete audit trail
- Cash allocation and margin management
-
Risk Management (
EventDriven/riskmanager/)- Greek-based limits (Delta, Gamma, Vega, Theta)
- Position analyzer with pluggable "cogs" for custom rules
- Intelligent order picker for option chain selection
- Dynamic position sizing (fixed, z-score-based, etc.)
-
Data Handling (
EventDriven/data.py,trade/datamanager/)- Historic trade data handler with bar-by-bar iteration
- Disk-based caching with configurable TTL (
CustomCache) - Integration with ThetaData, yfinance, OpenBB
- Market timeseries management with holiday/trading day awareness
-
Options Pricing Library (
trade/optionlib/)- Black-Scholes-Merton pricing and Greeks
- Implied volatility calculations
- Dividend schedule handling (discrete and continuous)
- Support for European and American options
-
Performance Analytics (
EventDriven/performance.py)- Sharpe ratio calculation
- Drawdown analysis
- Returns attribution
- Trade-level statistics
-
Configuration System (
EventDriven/configs/)- Centralized configuration management via dataclasses
- Frozen configs for immutability guarantees
- Validation and type checking
- Export/import configuration dictionaries
- Python: 3.10+ recommended (uses modern type hints)
- Key Dependencies: numpy, pandas, matplotlib, yfinance, QuantLib, diskcache
- Optional: ThetaData API access for high-quality options data
# Clone the repository
git clone https://github.com/Zino-ctrlZ/QuantTools.git
cd QuantTools
# Install in editable mode with dependencies
pip install -e .
# Verify installation
python -c "from EventDriven.backtest import OptionSignalBacktest; print('Success!')"Create a .env file in the repo root:
# Required paths
WORK_DIR=/path/to/QuantTools
DBASE_DIR=/path/to/your/database/module # if using ThetaData
# Optional: data sources
THETA_USERNAME=your_username
THETA_PASSWORD=your_password
# Cache configuration
GEN_CACHE_PATH=/path/to/cache/directoryQuantTools/
├── EventDriven/ # Core backtesting engine
│ ├── backtest.py # Main OptionSignalBacktest class
│ ├── new_portfolio.py # Portfolio state management
│ ├── strategy.py # Strategy base classes
│ ├── data.py # Data handlers (HistoricTradeDataHandler)
│ ├── event.py # Event types (MarketEvent, SignalEvent, etc.)
│ ├── eventScheduler.py # Event queue coordinator
│ ├── execution.py # Execution handlers with slippage
│ ├── trade.py # Trade object definitions
│ ├── tradeLedger.py # Trade ledger for audit trail
│ ├── performance.py # Performance metrics (Sharpe, drawdowns)
│ ├── helpers.py # Utility functions
│ ├── types.py # Type definitions and enums
│ ├── exceptions.py # Custom exceptions
│ ├── riskmanager/ # Risk management system
│ │ ├── new_base.py # RiskManager orchestrator
│ │ ├── order_picker.py # Option chain search logic
│ │ ├── cogs/ # Pluggable risk rules
│ │ └── market_data.py # Market data caching
│ ├── configs/ # Configuration system
│ │ ├── core.py # Config dataclasses
│ │ ├── base.py # Base config functionality
│ │ └── export_configs.py # Config serialization
│ ├── dataclasses/ # Structured data types
│ │ ├── orders.py # Order request structures
│ │ └── states.py # Position/portfolio states
│ ├── demos/ # Example scripts
│ │ └── demoRun.py # Full backtest example
│ └── notebooks/ # Jupyter exploration notebooks
│
├── trade/ # Trading utilities library
│ ├── helpers/ # Helper functions
│ │ ├── helper.py # Core utilities, CustomCache
│ │ ├── Logging.py # Logging setup
│ │ └── Configuration.py # Config management
│ ├── optionlib/ # Options pricing library
│ │ ├── pricing/ # Pricing models (BSM, etc.)
│ │ ├── greeks/ # Greek calculations
│ │ ├── vol/ # Volatility models
│ │ └── assets/ # Dividend handling
│ ├── assets/ # Asset classes (Stock, Option)
│ ├── datamanager/ # Data management (WIP on CHIDI-JAN04 branch)
│ ├── backtester_/ # Legacy backtester (being replaced)
│ └── models/ # Statistical/ML models
│
├── module_test/ # Module-level test scripts
├── setup.py # Package installation
├── setup.cfg # Package metadata
├── ruff.toml # Linting configuration (Ruff)
├── pricingConfig.json # Option pricing defaults
└── logs/ # Runtime logs
- EventDriven/: Complete event-driven backtesting framework. Import from here for backtests.
- trade/helpers/: Cross-cutting utilities (caching, logging, date helpers, config management).
- trade/optionlib/: Options pricing and Greeks. Used by RiskManager for valuation.
- trade/assets/: Asset class definitions (Stock, Option). Provides data access methods.
- EventDriven/riskmanager/: All risk logic (limits, sizing, order selection). Plugs into Portfolio.
- EventDriven/configs/: Configuration dataclasses. Pass these to control backtest behavior.
QuantTools processes trades chronologically through an event queue:
1. MarketEvent → 2. SignalEvent → 3. OrderEvent → 4. FillEvent
↓ ↓ ↓ ↓
DataHandler Strategy RiskManager/Portfolio Executor
Event Types (see EventDriven/event.py):
- MarketEvent: New bar available, triggers strategy to check for signals
- SignalEvent: Strategy wants to open/close a position
- OrderEvent: Portfolio approved an order, sends to executor
- FillEvent: Order filled, updates portfolio holdings
- RollEvent: Option position needs rolling (exercise, expiry)
- ExerciseEvent: Option exercised early (American options)
Workflow:
HistoricTradeDataHandleriterates through dates, emittingMarketEventStrategy.calculate_signals()analyzes current bar, emitsSignalEventPortfolio.analyze_signal()checks cash/limits, requests order fromRiskManagerRiskManagersearches option chains, returnsOrderRequest- Portfolio converts to
OrderEvent, puts in queue SimulatedExecutionHandlersimulates fill with slippage, emitsFillEventPortfolio.update_fill()updates positions and cash
CustomCache (trade/helpers/helper.py):
- Disk-backed cache using
diskcachelibrary - Configurable expiration (default 7 days)
- Used for market data, option chains, pricing calculations
- Example:
from trade.helpers.helper import CustomCache cache = CustomCache(location='/tmp/mycache', fname='test', expire_days=30) cache['AAPL_2024-01-01'] = {'close': 185.92} value = cache.get('AAPL_2024-01-01', default=None)
HistoricTradeDataHandler (EventDriven/data.py):
- Loads trade data from DataFrame
- Iterates bar-by-bar simulating real-time data feed
- Handles multi-symbol backtests
- Emits
MarketEventfor each new bar
Market Timeseries (EventDriven/riskmanager/market_data.py):
- Manages historical price data for underlyings
- Caches OHLCV data per symbol
- Provides dividend yield history
- Handles trading day calendars
OptionSignalPortfolio (EventDriven/new_portfolio.py):
- Maintains current positions (long/short options and equities)
- Tracks cash, holdings value, unrealized P&L
- Generates orders from signals (via RiskManager)
- Updates state on fills
- Manages position lifecycle (open, roll, close, exercise)
RiskManager (EventDriven/riskmanager/new_base.py):
- OrderPicker: Searches option chains for contracts matching criteria (delta, strike, expiry)
- PositionAnalyzer: Runs "cogs" (modular rules) on current positions to recommend actions
- LimitsAndSizing: Enforces Greek limits, position size constraints
- Market Data: Caches option chains, spot prices, dividends
Cogs (EventDriven/riskmanager/cogs/):
Pluggable modules that analyze positions and return recommendations:
LimitsAndSizingCog: Check if position violates Greek/size limitsPositionSignalsCog: Re-evaluate signals for open positionsStrategyRollCog: Determine if positions should be rolled- Custom cogs: Extend
BaseCogfor your own logic
Configurations use frozen dataclasses (EventDriven/configs/core.py):
from EventDriven.configs.core import BacktesterConfig, RiskManagerConfig
# Configure backtest behavior
backtest_config = BacktesterConfig(
t_plus_n=1, # T+1 settlement
max_slippage_pct=0.0015, # 0.15% slippage
slippage_enabled=True,
logger_override_level="INFO"
)
# Configure risk controls
risk_config = RiskManagerConfig(
delta_limit=0.30, # Max 30% delta exposure per position
gamma_limit=0.10,
vega_limit=0.50,
use_portfolio_level_limits=True
)Pass configs to classes:
backtest = OptionSignalBacktest(trades_df, config=backtest_config)
risk_manager = RiskManager(bars, events, config=risk_config)Logging uses Python's standard logging module configured via trade/helpers/Logging.py:
from trade.helpers.Logging import setup_logger
logger = setup_logger('MyModule', log_level='DEBUG')
logger.info("Backtest started")
logger.warning("Position limit exceeded")
logger.error("Failed to retrieve option chain", exc_info=True)Logs are written to:
- Console (stdout/stderr)
logs/__main__.log.<date>(daily rotation)
Linting: Uses ruff (configured in ruff.toml):
ruff check . # Check for issues
ruff check --fix . # Auto-fix issues
ruff format . # Format codeThis example creates a simple backtest with dummy trade data:
import pandas as pd
from EventDriven.backtest import OptionSignalBacktest
from EventDriven.configs.core import BacktesterConfig
# Create sample trade data
trades = pd.DataFrame({
'Ticker': ['AAPL', 'AAPL', 'MSFT'],
'EntryTime': ['2024-01-02', '2024-01-10', '2024-01-05'],
'ExitTime': ['2024-01-15', '2024-01-25', '2024-01-20'],
'Size': [10, -5, 8], # Positive = long, negative = short
'EntryPrice': [150.0, 155.0, 380.0],
'ExitPrice': [158.0, 152.0, 390.0],
'Type': ['CALL', 'PUT', 'CALL'],
'Strike': [145, 160, 375],
'Expiry': ['2024-02-16', '2024-02-16', '2024-02-16']
})
# Configure backtest
config = BacktesterConfig(
t_plus_n=1, # T+1 settlement
max_slippage_pct=0.001, # 0.1% slippage
slippage_enabled=True
)
# Run backtest
backtest = OptionSignalBacktest(
trades=trades,
initial_capital=100000,
config=config
)
# Execute
import asyncio
asyncio.run(backtest.run())
# Access results
print(f"Final Portfolio Value: ${backtest.portfolio.current_holdings['total']:,.2f}")
print(f"Total Return: {backtest.portfolio.current_holdings['total'] / 100000 - 1:.2%}")
# View trade ledger
ledger = backtest.portfolio.ledger
print("\nTrade Ledger:")
print(ledger[['date', 'ticker', 'action', 'quantity', 'price', 'cost']])
# Performance metrics
from EventDriven.performance import create_sharpe_ratio, create_drawdowns
equity_curve = backtest.portfolio.all_holdings
returns = equity_curve['total'].pct_change().dropna()
sharpe = create_sharpe_ratio(returns, periods=252)
print(f"\nSharpe Ratio: {sharpe:.3f}")Expected Output Shape:
portfolio.ledger: DataFrame with columns['date', 'ticker', 'action', 'quantity', 'price', 'cost', 'commission']portfolio.all_holdings: DataFrame with datetime index, columns for each symbol + 'cash', 'commission', 'total'portfolio.current_positions: Dict mapping position_id → PositionState objects- Logs printed to console showing event processing
Enforce Greek limits and position sizing constraints:
from EventDriven.backtest import OptionSignalBacktest
from EventDriven.configs.core import BacktesterConfig, RiskManagerConfig
# Configure tight risk controls
risk_config = RiskManagerConfig(
delta_limit=0.25, # Max 25% delta per position
gamma_limit=0.08, # Limit gamma exposure
vega_limit=0.40, # Limit vega exposure
max_position_size=20, # Max 20 contracts per position
use_portfolio_level_limits=True
)
backtest_config = BacktesterConfig(
t_plus_n=1,
risk_manager_config=risk_config
)
# Run backtest - orders violating limits will be rejected
backtest = OptionSignalBacktest(
trades=your_trades_df,
initial_capital=100000,
config=backtest_config
)
await backtest.run()
# Check rejected orders
print("Rejected Orders:", backtest.portfolio.ledger[backtest.portfolio.ledger['action'] == 'REJECTED'])Assumptions: Requires your_trades_df with columns: ['Ticker', 'EntryTime', 'ExitTime', 'Size', 'EntryPrice', 'ExitPrice', 'Type', 'Strike', 'Expiry']
Cache expensive computations (e.g., option chains, pricing data):
from trade.helpers.helper import CustomCache
import pandas as pd
# Initialize cache with 30-day expiration
cache = CustomCache(
location='/tmp/options_cache',
fname='aapl_chains',
expire_days=30,
clear_on_exit=False # Persist across runs
)
# Cache option chain data
ticker = 'AAPL'
date = '2024-01-02'
key = f"{ticker}_{date}_chain"
if key in cache:
chain = cache[key]
print("Loaded from cache")
else:
# Expensive operation: fetch option chain
chain = fetch_option_chain(ticker, date) # Your data source
cache[key] = chain
print("Cached for future use")
# Use the chain
print(chain[['strike', 'call_bid', 'call_ask', 'iv']])Notes:
- Cache automatically cleans up expired entries
- Set
clear_on_exit=Truefor temporary caches (testing) - Cache location persists; same
fnameretrieves same cache across sessions
Directly interact with Portfolio for custom workflows:
from EventDriven.new_portfolio import OptionSignalPortfolio
from EventDriven.data import HistoricTradeDataHandler
from EventDriven.eventScheduler import EventScheduler
from EventDriven.riskmanager.new_base import RiskManager
from EventDriven.event import FillEvent
from EventDriven.types import FillDirection
# Setup components
bars = HistoricTradeDataHandler(trades_df, symbol_list=['AAPL'])
events = EventScheduler()
risk_mgr = RiskManager(bars, events)
portfolio = OptionSignalPortfolio(
bars=bars,
eventScheduler=events,
risk_manager=risk_mgr,
initial_capital=50000
)
# Manually create a fill (simulating order execution)
fill = FillEvent(
timeindex=pd.Timestamp('2024-01-02'),
symbol='AAPL',
exchange='NASDAQ',
quantity=10,
direction=FillDirection.BUY,
fill_cost=1500.0, # Total cost including commission
commission=1.50,
option_type='CALL',
strike=150,
expiry='2024-02-16'
)
# Update portfolio with fill
portfolio.update_fill(fill)
# Check positions
print("Current Positions:", portfolio.current_positions)
print("Current Cash:", portfolio.current_holdings['cash'])Use Case: Manual portfolio construction for sensitivity analysis or custom order sequences.
Create a strategy that generates signals based on your logic:
from EventDriven.strategy import Strategy
from EventDriven.event import SignalEvent
from EventDriven.types import SignalTypes
from EventDriven.helpers import generate_signal_id
class MomentumStrategy(Strategy):
"""
Simple momentum strategy: buy when 20-day MA > 50-day MA
"""
def __init__(self, bars, events, short_window=20, long_window=50):
self.bars = bars
self.events = events
self.symbol_list = bars.symbol_list
self.short_window = short_window
self.long_window = long_window
self.positions = {s: False for s in self.symbol_list}
def calculate_signals(self, event):
if event.type != 'MARKET':
return
for symbol in self.symbol_list:
bars = self.bars.get_latest_bars(symbol, N=self.long_window)
if len(bars) < self.long_window:
continue
# Calculate moving averages
closes = [bar['close'] for bar in bars]
short_ma = sum(closes[-self.short_window:]) / self.short_window
long_ma = sum(closes) / self.long_window
# Generate signal
if short_ma > long_ma and not self.positions[symbol]:
# Buy signal
signal = SignalEvent(
ticker=symbol,
datetime=bars[-1]['datetime'],
signal_id=generate_signal_id(),
signal_type=SignalTypes.LONG,
suggested_quantity=1
)
self.events.put(signal)
self.positions[symbol] = True
elif short_ma < long_ma and self.positions[symbol]:
# Sell signal
signal = SignalEvent(
ticker=symbol,
datetime=bars[-1]['datetime'],
signal_id=generate_signal_id(),
signal_type=SignalTypes.EXIT,
suggested_quantity=-1
)
self.events.put(signal)
self.positions[symbol] = False
# Use in backtest
from EventDriven.backtest import OptionSignalBacktest
# Replace default strategy with custom one
backtest = OptionSignalBacktest(trades_df, initial_capital=100000)
backtest.strategy = MomentumStrategy(backtest.bars, backtest.eventScheduler)
await backtest.run()Note: Custom strategies must inherit from Strategy and implement calculate_signals(event).
Use RiskManager to evaluate Greeks for current positions:
from EventDriven.riskmanager.new_base import RiskManager
from EventDriven.configs.core import RiskManagerConfig
# Configure Greek limits
config = RiskManagerConfig(
delta_limit=0.30,
gamma_limit=0.10,
vega_limit=0.50,
theta_limit=-0.05 # Negative = losing value over time
)
risk_mgr = RiskManager(bars, events, config=config)
# Analyze a specific position
position_state = portfolio.current_positions['AAPL_CALL_150_2024-02-16']
# Load market data for analysis
analysis_date = pd.Timestamp('2024-01-15')
risk_mgr._load_market_data_for_date(analysis_date)
# Calculate Greeks (done automatically in analyze_position)
from EventDriven.dataclasses.states import PositionAnalysisContext
context = PositionAnalysisContext(
position=position_state,
analysis_date=analysis_date,
portfolio_state=portfolio.get_portfolio_state(),
market_data=risk_mgr.timeseries
)
# Run analysis (checks limits, signals, rolls)
recommendations = risk_mgr.analyze_position(context)
print("Position Analysis:")
print(f"Delta: {position_state.greeks.delta:.3f} (Limit: {config.delta_limit})")
print(f"Gamma: {position_state.greeks.gamma:.3f} (Limit: {config.gamma_limit})")
print(f"Vega: {position_state.greeks.vega:.3f} (Limit: {config.vega_limit})")
print(f"Recommendations: {recommendations}")Use Case: Real-time position monitoring, pre-trade risk checks, limit breach alerts.
QuantTools supports both discrete and continuous dividend models:
from trade.optionlib.assets.dividend import (
get_vectorized_dividend_schedule,
get_vectorized_continuous_dividends
)
from trade.optionlib.config.types import DiscreteDivGrowthModel
# Discrete dividends (actual ex-dates and amounts)
div_schedule = get_vectorized_dividend_schedule(
tickers=['AAPL'],
start_dates=['2024-01-01'],
end_dates=['2024-12-31'],
method=DiscreteDivGrowthModel.CONSTANT_AVG.value,
lookback_years=2
)
print("Discrete Dividend Schedule:")
for entry in div_schedule[0].schedule:
print(f" {entry.date}: ${entry.amount:.2f}")
# Continuous dividend yield (for European options)
continuous_yield = get_vectorized_continuous_dividends(
tickers=['AAPL'],
start_dates=['2024-01-01'],
end_dates=['2024-12-31']
)
print(f"\nAnnualized Dividend Yield: {continuous_yield[0]:.2%}")
# Use in option pricing
from trade.optionlib.pricing.bsm import black_scholes_merton_price
price = black_scholes_merton_price(
S=150.0, # Spot price
K=145.0, # Strike
T=0.5, # Time to expiry (years)
r=0.05, # Risk-free rate
sigma=0.25, # Implied volatility
q=continuous_yield[0], # Dividend yield
option_type='call'
)
print(f"Option Price: ${price:.2f}")Assumptions: Requires dividend history data (fetched from yfinance or database).
Tip 1: Cache Option Chains Aggressively
# BAD: Fetching chains every iteration
for date in dates:
chain = fetch_option_chain(ticker, date) # Network call every time
# GOOD: Cache chains with CustomCache
cache = CustomCache(fname='option_chains', expire_days=7)
for date in dates:
key = f"{ticker}_{date}"
if key not in cache:
cache[key] = fetch_option_chain(ticker, date)
chain = cache[key]Tip 2: Use T+N Settlement Delays
# BAD: Assumes instant settlement
config = BacktesterConfig(t_plus_n=0) # Unrealistic
# GOOD: T+1 for options/equities
config = BacktesterConfig(t_plus_n=1) # Matches real settlementTip 3: Handle Missing Data Gracefully
# BAD: Crash on missing bar
bars = self.bars.get_latest_bars(symbol, N=50)
close = bars[-1]['close'] # IndexError if bars empty
# GOOD: Check before accessing
bars = self.bars.get_latest_bars(symbol, N=50)
if bars is None or len(bars) < 50:
return # Skip this symbol
close = bars[-1]['close']Pitfall 1: Forgetting to Update Market Data
When using RiskManager standalone, call _load_market_data_for_date() before analysis:
risk_mgr._load_market_data_for_date(analysis_date)
recommendations = risk_mgr.analyze_position(context)Pitfall 2: Mutating Shared State
Portfolio state is mutable. Use deepcopy if needed:
from copy import deepcopy
current_state = deepcopy(portfolio.current_positions)
# Modify current_state without affecting portfolioPitfall 3: Mixing Async and Sync Code
OptionSignalBacktest.run() is async. Always use await or asyncio.run():
# BAD
backtest.run() # Returns coroutine, doesn't execute
# GOOD
import asyncio
asyncio.run(backtest.run())
# Or in Jupyter
await backtest.run()Use the vectorized Black-Scholes implementation for fast option pricing:
from trade.optionlib.pricing.black_scholes import (
black_scholes_vectorized,
black_scholes_with_carry_div
)
from trade.optionlib.assets.forward import EquityForward
import numpy as np
# Single option pricing
spot = 150.0
strike = 155.0
time_to_expiry = 0.25 # 3 months
risk_free_rate = 0.05
volatility = 0.30
dividend_yield = 0.02
# Calculate forward price
forward = EquityForward(
S=spot,
T=time_to_expiry,
r=risk_free_rate,
q=dividend_yield
)
F = forward.price()
# Price call and put
call_price = black_scholes_vectorized(
F=F,
K=strike,
T=time_to_expiry,
r=risk_free_rate,
sigma=volatility,
option_type='c'
)
put_price = black_scholes_vectorized(
F=F,
K=strike,
T=time_to_expiry,
r=risk_free_rate,
sigma=volatility,
option_type='p'
)
print(f"Call Price: ${call_price[0]:.2f}")
print(f"Put Price: ${put_price[0]:.2f}")
# Vectorized pricing across multiple strikes
strikes = np.array([145, 150, 155, 160, 165])
call_prices = black_scholes_vectorized(
F=F,
K=strikes,
T=time_to_expiry,
r=risk_free_rate,
sigma=volatility,
option_type='c'
)
import pandas as pd
chain = pd.DataFrame({
'Strike': strikes,
'Call': call_prices,
'Intrinsic': np.maximum(spot - strikes, 0),
'Time_Value': call_prices - np.maximum(spot - strikes, 0)
})
print("\nOption Chain:")
print(chain)Output: Vectorized pricing is ~10x faster than looping for large chains.
Greeks can be calculated analytically or numerically:
from trade.optionlib.greeks import vectorized_market_greeks_bsm
from trade.optionlib.greeks.numerical.finite_diff import FiniteGreeksEstimator
# Analytical Greeks (fast, exact for BSM)
greeks_result = vectorized_market_greeks_bsm(
S=150.0,
K=155.0,
T=0.25,
r=0.05,
sigma=0.30,
q=0.02,
option_type='c'
)
print("Analytical Greeks (BSM):")
print(f" Delta: {greeks_result.delta[0]:.4f}")
print(f" Gamma: {greeks_result.gamma[0]:.4f}")
print(f" Vega: {greeks_result.vega[0]:.4f}")
print(f" Theta: {greeks_result.theta[0]:.4f}")
print(f" Rho: {greeks_result.rho[0]:.4f}")
# Numerical Greeks (for American options or complex models)
from trade.optionlib.pricing.bjs2002 import bjerksund_stensland_2002_vectorized
estimator = FiniteGreeksEstimator(
price_function=bjerksund_stensland_2002_vectorized,
params={
'S': 150.0,
'K': 155.0,
'T': 0.25,
'r': 0.05,
'sigma': 0.30,
'q': 0.02,
'option_type': 'c'
}
)
# Calculate first-order Greeks
delta = estimator.delta()
vega = estimator.vega()
theta = estimator.theta()
# Calculate second-order Greeks
gamma = estimator.gamma()
vomma = estimator.vomma() # d²V/dσ²
vanna = estimator.vanna() # d²V/dS/dσ
print("\nNumerical Greeks (American Option):")
print(f" Delta: {delta:.4f}")
print(f" Gamma: {gamma:.4f}")
print(f" Vega: {vega:.4f}")
print(f" Vomma: {vomma:.6f}")
print(f" Vanna: {vanna:.6f}")Use Case: Risk management, hedging, Greeks-based trading strategies.
Extract implied volatility from market prices:
from trade.optionlib.vol.implied_vol import (
bsm_vol_est_minimization,
bsm_vol_est_brute_force,
vectorized_iv_calculation
)
# Single IV calculation
market_price = 5.50
spot = 150.0
strike = 155.0
time_to_expiry = 0.25
risk_free_rate = 0.05
dividend_yield = 0.02
# Calculate forward
from trade.optionlib.assets.forward import EquityForward
forward = EquityForward(S=spot, T=time_to_expiry, r=risk_free_rate, q=dividend_yield)
F = forward.price()
# Method 1: Optimization-based (fast, accurate)
iv_optimized = bsm_vol_est_minimization(
F=F,
K=strike,
T=time_to_expiry,
r=risk_free_rate,
market_price=market_price,
option_type='c'
)
print(f"Implied Volatility (Optimized): {iv_optimized:.2%}")
# Method 2: Brute force (slower but robust)
iv_brute = bsm_vol_est_brute_force(
F=F,
K=strike,
T=time_to_expiry,
r=risk_free_rate,
market_price=market_price,
option_type='c'
)
print(f"Implied Volatility (Brute Force): {iv_brute:.2%}")
# Vectorized IV calculation for entire chain
import numpy as np
strikes = np.array([145, 150, 155, 160, 165])
market_prices = np.array([8.50, 5.80, 3.90, 2.50, 1.60])
ivs = vectorized_iv_calculation(
S=spot,
K=strikes,
T=time_to_expiry,
r=risk_free_rate,
q=dividend_yield,
market_prices=market_prices,
option_type='c'
)
# Plot volatility smile
import matplotlib.pyplot as plt
plt.figure(figsize=(10, 6))
plt.plot(strikes, ivs * 100, marker='o')
plt.xlabel('Strike Price')
plt.ylabel('Implied Volatility (%)')
plt.title('Volatility Smile')
plt.grid(True)
plt.show()Note: If market price is below intrinsic value, IV calculation will fail (returns NaN).
QuantTools includes powerful decorators for logging, timing, and profiling:
from trade.helpers.decorators import timeit, log_time, log_error, cProfiler
# @timeit: Track execution time and save metadata
@timeit
def expensive_computation(n):
"""Simulate expensive computation"""
result = sum(i**2 for i in range(n))
return result
# Run multiple times
for _ in range(10):
expensive_computation(1000000)
# Timing data is automatically saved to .cache/timeit_log.csv
# Columns: date, timestamp, func_name, module, duration, args, kwargs
# @log_time: Log execution time with custom logger
from trade.helpers.Logging import setup_logger
logger = setup_logger('MyModule')
@log_time(logger=logger)
def fetch_market_data(ticker, start, end):
"""Simulated data fetch"""
import time
time.sleep(0.5)
return f"Data for {ticker} from {start} to {end}"
result = fetch_market_data('AAPL', '2024-01-01', '2024-12-31')
# Logs: "fetch_market_data took 0.5012s"
# @log_error: Automatic error logging with stack traces
@log_error(logger=logger, raise_exception=True)
def risky_calculation(x, y):
return x / y # Will crash if y=0
try:
risky_calculation(10, 0)
except ZeroDivisionError:
print("Error was logged automatically")
# @cProfiler: Detailed profiling
@cProfiler
def complex_backtest():
# Your backtest code here
pass
# Generates profile.prof file for analysis with snakeviz or pstatsBest Practice: Use @timeit in production to track performance regressions.
The Context manager handles time windows for data fetching:
from trade.helpers.Context import Context
# Example 1: Fetch 1 year of daily data ending today
with Context(timewidth='1', timeframe='year', print_context=True):
# Inside this context, all data fetches use these settings
from trade.assets.Stock import Stock
stock = Stock('AAPL')
data = stock.spot(ts=True) # Uses context settings
print(f"Fetched {len(data)} days of data")
# Example 2: Fetch specific date range
with Context(start_date='2024-01-01', end_date='2024-12-31'):
stock = Stock('MSFT')
data = stock.spot(ts=True)
print(f"Date range: {data.index.min()} to {data.index.max()}")
# Example 3: Intraday data
with Context(
start_date='2024-01-02 09:30',
end_date='2024-01-02 16:00',
timeframe='minute',
timewidth='1'
):
stock = Stock('GOOGL')
intraday = stock.spot(ts=True, interval='1m')
print(f"Fetched {len(intraday)} minute bars")
# Context automatically:
# - Validates dates (no weekends/holidays)
# - Adjusts to last business day if needed
# - Sets default times (9:30 AM start, 4:00 PM end)
# - Configures global settings accessed by Stock/Option classesUse Case: Consistent time handling across multiple data sources.
The trade.backtester_ module provides a fast vectorized backtester:
from trade.backtester_.backtester_ import PTBacktester
from trade.backtester_.data import PTDataset
from trade.backtester_._strategy import StrategyBase
import pandas as pd
import yfinance as yf
# Define a simple moving average strategy
class SimpleMAStrategy(StrategyBase):
# Strategy parameters (can be optimized)
fast_ma = 20
slow_ma = 50
def init(self):
"""Initialize indicators"""
self.fast_sma = self.I(lambda: self.data.Close.rolling(self.fast_ma).mean())
self.slow_sma = self.I(lambda: self.data.Close.rolling(self.slow_ma).mean())
def next(self):
"""Trading logic called on each bar"""
if self.fast_sma[-1] > self.slow_sma[-1] and self.fast_sma[-2] <= self.slow_sma[-2]:
# Golden cross: buy
if not self.position:
self.buy()
elif self.fast_sma[-1] < self.slow_sma[-1] and self.fast_sma[-2] >= self.slow_sma[-2]:
# Death cross: sell
if self.position:
self.sell()
# Fetch data
tickers = ['AAPL', 'MSFT', 'GOOGL']
datasets = []
for ticker in tickers:
df = yf.download(ticker, start='2023-01-01', end='2024-12-31', interval='1d', progress=False)
df.columns = [col.capitalize() for col in df.columns] # Backtester expects capitalized columns
datasets.append(PTDataset(ticker, df))
# Run backtest
backtest = PTBacktester(
datalist=datasets,
strategy=SimpleMAStrategy,
cash=100000,
commission=0.001 # 0.1% commission
)
# Execute
stats = backtest.run()
# View results
print("\nBacktest Results:")
print(stats)
# Access detailed metrics
print(f"\nTotal Return: {stats['Return [%]'].mean():.2f}%")
print(f"Sharpe Ratio: {stats['Sharpe Ratio'].mean():.2f}")
print(f"Max Drawdown: {stats['Max. Drawdown [%]'].mean():.2f}%")
# Get trade history
trades = backtest.__trades()
print(f"\nTotal Trades: {len(trades)}")
print(trades[['Ticker', 'EntryTime', 'ExitTime', 'Size', 'ReturnPct']].head())
# Optimize parameters
from trade.backtester_.utils.utils import optimize
optimized_stats = optimize(
backtest=backtest,
params={
'fast_ma': range(10, 30, 5),
'slow_ma': range(40, 80, 10)
},
maximize='Sharpe Ratio',
max_tries=50
)
print(f"\nOptimal Parameters:")
print(f" Fast MA: {optimized_stats['fast_ma']}")
print(f" Slow MA: {optimized_stats['slow_ma']}")
print(f" Best Sharpe: {optimized_stats['Sharpe Ratio']:.2f}")Performance: PTBacktester is ~100x faster than event-driven for simple strategies but less realistic (no slippage, T+N delays).
Build and query volatility surfaces for options:
from trade.models.VolSurface import (
SurfaceBuilder,
SurfaceManager,
DumasModelBuilder
)
import pandas as pd
# Sample option chain data (from market)
chain_data = pd.DataFrame({
'strike': [140, 145, 150, 155, 160, 165, 170],
'dte': [30, 30, 30, 30, 30, 30, 30],
'implied_vol': [0.28, 0.26, 0.24, 0.25, 0.27, 0.29, 0.31],
'right': ['C', 'C', 'C', 'C', 'C', 'C', 'C']
})
spot_price = 150.0
# Build SVI (Stochastic Volatility Inspired) surface
surface_builder = SurfaceBuilder(
chain_data=chain_data,
spot=spot_price,
model_type='SVI'
)
# Fit the model
surface_builder.fit()
# Query implied volatility for arbitrary strikes/DTE
query_strikes = [147, 152, 158]
query_dte = 30
predicted_ivs = surface_builder.predict(
strikes=query_strikes,
dte=query_dte
)
print("Predicted Implied Volatilities:")
for strike, iv in zip(query_strikes, predicted_ivs):
print(f" Strike {strike}: {iv:.2%}")
# Dumas model (rolling regression for IV surface)
dumas_model = DumasModelBuilder(
chain_data=chain_data,
spot=spot_price,
rolling_window=7 # Days
)
dumas_model.fit()
# Predict IV for out-of-sample strikes
oos_strikes = [142, 168]
oos_predictions = dumas_model.predict(strikes=oos_strikes, dte=30)
print("\nDumas Model Predictions:")
for strike, iv in zip(oos_strikes, oos_predictions):
print(f" Strike {strike}: {iv:.2%}")
# Visualize surface
surface_builder.plot_surface(
strike_range=(130, 170),
dte_range=(7, 90)
)Use Case: Interpolate IVs for strikes not quoted, extrapolate to different DTEs, detect arbitrage opportunities.
Extend pandas-ta with custom indicators:
from trade.helpers.custom_ta import (
atr_trailing_stop,
wilders_average,
hull_average
)
import pandas as pd
import yfinance as yf
# Fetch data
df = yf.download('AAPL', start='2024-01-01', end='2024-12-31', progress=False)
# ATR Trailing Stop (for stop-loss calculation)
atr_stop = atr_trailing_stop(
df=df,
period=14,
multiplier=2.0,
trend_col='Close' # Column to base trend on
)
df['ATR_Stop_Long'] = atr_stop['long_stop']
df['ATR_Stop_Short'] = atr_stop['short_stop']
df['Trend'] = atr_stop['trend'] # 1 = uptrend, -1 = downtrend
print("ATR Trailing Stop:")
print(df[['Close', 'ATR_Stop_Long', 'Trend']].tail())
# Wilder's smoothing (for RSI, ATR)
wilders = wilders_average(df['Close'], length=14)
print(f"\nWilder's Average (14): {wilders.iloc[-1]:.2f}")
# Hull Moving Average (reduced lag)
hull = hull_average(df['Close'], length=20)
df['HMA_20'] = hull
# Plot
import matplotlib.pyplot as plt
plt.figure(figsize=(12, 6))
plt.plot(df.index, df['Close'], label='Close', alpha=0.7)
plt.plot(df.index, df['HMA_20'], label='HMA(20)', linewidth=2)
plt.plot(df.index, df['ATR_Stop_Long'], label='ATR Stop (Long)', linestyle='--')
plt.legend()
plt.title('Custom Indicators')
plt.show()Integration: Use these in StrategyBase or Strategy classes for signals.
Speed up data fetching and computations:
from trade.helpers.threads import runThreads
# Example: Fetch option chains for multiple underlyings
def fetch_option_chain(ticker, date):
"""Fetch option chain from data source"""
# Your data fetching logic
print(f"Fetching chain for {ticker} on {date}")
import time
time.sleep(0.5) # Simulate network call
return {'ticker': ticker, 'date': date, 'contracts': 100}
# Sequential (slow)
import time
tickers = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'META']
date = '2024-01-15'
start = time.time()
results_seq = [fetch_option_chain(t, date) for t in tickers]
print(f"Sequential: {time.time() - start:.2f}s")
# Parallel with runThreads (fast)
inputs = [[ticker, date] for ticker in tickers]
start = time.time()
results_parallel = runThreads(
func=fetch_option_chain,
OrderedInputs=inputs,
max_workers=5,
return_results=True
)
print(f"Parallel: {time.time() - start:.2f}s")
# Example 2: Parallel option pricing
from trade.optionlib.pricing.black_scholes import black_scholes_vectorized
from trade.optionlib.assets.forward import EquityForward
def price_option_with_greeks(strike, spot, T, r, sigma, q):
F = EquityForward(S=spot, T=T, r=r, q=q).price()
price = black_scholes_vectorized(F=F, K=strike, T=T, r=r, sigma=sigma, option_type='c')
return {'strike': strike, 'price': price[0]}
strikes = range(100, 200, 5)
inputs = [[k, 150.0, 0.25, 0.05, 0.30, 0.02] for k in strikes]
prices = runThreads(
func=price_option_with_greeks,
OrderedInputs=inputs,
max_workers=10,
return_results=True
)
print("\nOption Prices:")
for result in prices[:5]:
print(f" Strike {result['strike']}: ${result['price']:.2f}")Performance: ~5x speedup for I/O-bound tasks (data fetching), ~2x for CPU-bound (if GIL-limited).
Centralized configuration management:
from trade.helpers.Configuration import ConfigProxy, initialize_configuration
# Initialize global config
initialize_configuration()
# Access config (singleton pattern)
config = ConfigProxy()
# Read current settings
print(f"Current timeframe: {config.timeframe}")
print(f"Start date: {config.start_date}")
print(f"End date: {config.end_date}")
# Modify settings (affects all subsequent data fetches)
config.timeframe = 'hour'
config.timewidth = '4'
# Context manager auto-configures and restores
from trade.helpers.Context import Context, clear_context
with Context(timewidth='1', timeframe='day', start_date='2024-01-01'):
print(f"Inside context: {config.timeframe}") # 'day'
print(f"Outside context: {config.timeframe}") # Back to 'hour'
# Clear all context settings
clear_context()
print(f"After clear: {config.timeframe}") # Default valueUse Case: Consistent settings across modules without passing parameters everywhere.
Decompose option P&L into Greeks components to understand what drives portfolio performance:
from datetime import datetime
from trade.assets.calculate.xmultiply_attr import (
load_option_pnl_data,
calculate_pnl_decomposition
)
from trade.assets.calculate.data_classes import TradePnlInfo
from EventDriven.types import PositionEffect
import pandas as pd
import matplotlib.pyplot as plt
# Define the option position
opttick = 'AAPL240621C00150000' # AAPL June 21, 2024 $150 Call
yesterday = datetime(2024, 3, 30)
today = datetime(2024, 3, 31)
# Load option P&L data (loads market data for Greeks and attribution)
payload = load_option_pnl_data(
yesterday=yesterday,
today=today,
opttick=opttick
)
# Calculate P&L decomposition
result = calculate_pnl_decomposition(payload)
# Access attribution results
attribution_df = result.attribution
print("\n=== P&L Attribution ===")
print(attribution_df)
# Columns in attribution_df:
# - delta_pnl: P&L from spot price movement
# - gamma_pnl: P&L from convexity (second-order spot effect)
# - theta_pnl: P&L from time decay
# - vega_pnl: P&L from volatility changes
# - volga_pnl: P&L from volatility convexity (second-order vol effect)
# - vanna_pnl: P&L from cross-effect of spot and vol
# - rho_pnl: P&L from interest rate changes
# - total_pnl_excl_trade_pnl: Sum of all Greek P&L components
# - unexplained_pnl: Difference between actual and attributed P&L
# - opt_dod_change: Observed option price change (day-over-day)
# Access underlying market data
print("\n=== Market Data ===")
print(f"Spot prices: {result.asset_payload.spot}")
print(f"Volatility: {result.vol}")
print(f"Greeks: {result.greeks}")
print(f"Day-over-day changes: {result.dod_change}")
# --- Advanced: Adjust for trade P&L (position entries/exits) ---
# When opening/closing positions, adjust attribution to exclude trade execution P&L
trade_info = TradePnlInfo(
position_effect_close=155.0, # Execution price
effect_date=datetime(2024, 3, 15),
tmin0_close=156.0, # Market close on T+0
tmin1_close=154.0, # Market close on T-1
position_effect=PositionEffect.OPEN,
quantity=10, # Number of contracts
position_entry_price=155.0
)
# Recalculate with trade adjustment
result_with_trade = calculate_pnl_decomposition(
payload,
trade_pnl_entries=[trade_info]
)
print("\n=== Attribution with Trade Adjustment ===")
print(result_with_trade.attribution)
# --- Visualize P&L Components ---
fig, axes = plt.subplots(2, 2, figsize=(15, 10))
# 1. Cumulative P&L by Greek
ax = axes[0, 0]
greeks = ['delta_pnl', 'gamma_pnl', 'theta_pnl', 'vega_pnl', 'volga_pnl']
attribution_df[greeks].cumsum().plot(ax=ax, title='Cumulative P&L by Greek')
ax.set_ylabel('Cumulative P&L ($)')
ax.legend(loc='best')
ax.grid(True, alpha=0.3)
# 2. Daily attribution waterfall
ax = axes[0, 1]
latest = attribution_df.iloc[-1]
components = ['delta_pnl', 'gamma_pnl', 'theta_pnl', 'vega_pnl', 'volga_pnl', 'vanna_pnl', 'rho_pnl']
values = [latest[c] for c in components]
ax.bar(range(len(components)), values, color=['green' if v > 0 else 'red' for v in values])
ax.set_xticks(range(len(components)))
ax.set_xticklabels([c.replace('_pnl', '').title() for c in components], rotation=45)
ax.set_title(f"Latest Day Attribution ({attribution_df.index[-1].date()})")
ax.set_ylabel('P&L Contribution ($)')
ax.axhline(0, color='black', linewidth=0.8)
ax.grid(True, alpha=0.3, axis='y')
# 3. Actual vs Total Attribution
ax = axes[1, 0]
attribution_df[['opt_dod_change', 'total_pnl_excl_trade_pnl']].cumsum().plot(
ax=ax, title='Actual vs Attributed P&L'
)
ax.set_ylabel('Cumulative P&L ($)')
ax.legend(['Actual P&L', 'Total Attribution'])
ax.grid(True, alpha=0.3)
# 4. Unexplained P&L (model error)
ax = axes[1, 1]
attribution_df['unexplained_pnl'].cumsum().plot(
ax=ax, color='orange', title='Cumulative Unexplained P&L'
)
ax.set_ylabel('Unexplained P&L ($)')
ax.axhline(0, color='black', linewidth=0.8)
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.show()
# --- Batch Processing: Multiple Days ---
from pandas.tseries.offsets import BDay
start_date = datetime(2024, 1, 1)
end_date = datetime(2024, 3, 31)
# Generate business day range
date_range = pd.date_range(start=start_date, end=end_date, freq=BDay())
all_attributions = []
for i in range(1, len(date_range)):
yesterday = date_range[i-1]
today = date_range[i]
payload = load_option_pnl_data(yesterday, today, opttick)
result = calculate_pnl_decomposition(payload)
all_attributions.append(result.attribution)
# Combine all periods
full_attribution = pd.concat(all_attributions)
print("\n=== Full Period Attribution ===")
print(full_attribution)
# --- Use Case: Risk Factor Analysis ---
# Identify top P&L contributors
total_attribution = attribution_data.sum()
top_contributors = total_attribution[greeks].abs().sort_values(ascending=False)
print("\n=== Top Risk Factor Contributors ===")
for greek, value in top_contributors.items():
pct = (value / attribution_data['Actual_PnL'].sum().abs()) * 100
print(f"{greek.replace('_PnL', '')}: ${value:.2f} ({pct:.1f}%)")
# Days with large unexplained P&L (potential model issues)
large_unexplained = attribution_data[attribution_data['Unexplained_PnL'].abs() > 10]
if len(large_unexplained) > 0:
print(f"\n⚠️ {len(large_unexplained)} days with |Unexplained P&L| > $10")
print(large_unexplained[['Actual_PnL', 'Total_PnL', 'Unexplained_PnL']])Key Insights:
-
xMULTIPLY Attribution Model: Uses Taylor expansion with Greeks from previous day
- Delta: Linear spot exposure (first-order)
- Gamma: Convexity from spot moves (second-order)
- Vega: Linear volatility exposure
- Volga: Convexity from vol moves (second-order)
- Vanna: Cross-effect between spot and vol
- Theta: Time decay
- Rho: Interest rate sensitivity
-
Unexplained P&L Sources:
- Jump risk (overnight gaps)
- Volatility smile/skew dynamics (not captured by flat vol)
- Execution slippage vs mid-market prices
- Data quality issues (missing/stale data)
- Higher-order Greeks not included (charm, vomma, etc.)
- Discrete daily rebalancing vs continuous model
-
Trade P&L Adjustment:
- Use
TradePnlInfoto exclude trade execution P&L from attribution - OPEN positions: Adjusts for difference between entry price and T-1 close
- CLOSE positions: Captures realized P&L on exit
- Keeps attribution focused on market risk factors, not execution
- Use
-
Data Requirements:
- Option tick format:
TICKER<YYMMDD><C/P><STRIKE> - Spot prices: Retrieved from market data
- Volatility: Implied from option prices
- Greeks: Calculated from DataManager
- Risk-free rates: USD rates from Treasury data
- Option tick format:
-
Use Cases:
- Performance Attribution: Decompose which Greeks drove portfolio returns
- Risk Management: Identify dominant risk exposures
- Strategy Validation: Verify strategy P&L matches expected Greek exposures
- Model Validation: Monitor unexplained P&L for model accuracy
- Trade Post-Mortem: Analyze individual position P&L drivers
Pro Tip: For portfolio-level attribution, run calculate_pnl_decomposition for each position and aggregate the results. This gives you total delta P&L, gamma P&L, etc., across all holdings. High unexplained P&L (>5% of total) indicates model issues or missing risk factors
large_unexplained = full_attribution[full_attribution['unexplained_pnl'].abs() > 10]
if len(large_unexplained) > 0:
print(f"\n
Additional sample code and notebooks demonstrating specific workflows:
- Full Backtest Workflow: See
EventDriven/demos/demoRun.pyfor a complete example using real data - Risk Manager Deep Dive: Explore
EventDriven/riskmanager/submodules for advanced risk controls - Option Pricing Examples: See
trade/optionlib/notebooks/for pricing and Greek calculations - Portfolio Construction: Check
EventDriven/notebooks/for portfolio analysis examples - Data Management: See
trade/datamanager/notebooks/create.ipynbfor DataManager patterns (WIP on CHIDI-JAN04 branch)
Error: ModuleNotFoundError: No module named 'EventDriven'
- Fix: Install package in editable mode:
pip install -e .from repo root - Check: Ensure you're in a venv/conda env with dependencies installed
Error: ModuleNotFoundError: No module named 'dbase'
- Context: Some modules import from
dbase(separate database package for ThetaData) - Fix Option 1: Install
dbasepackage if you have it - Fix Option 2: Comment out ThetaData imports if not using that data source
- Fix Option 3: Use yfinance data source instead (no external DB needed)
Error: ImportError: cannot import name 'query_database' from 'dbase.database.SQLHelpers'
- Fix: Set
DBASE_DIRin.envor remove ThetaData-specific code
Error: KeyError: 'WORK_DIR' or KeyError: 'GEN_CACHE_PATH'
- Fix: Create
.envfile with required paths:WORK_DIR=/path/to/QuantTools GEN_CACHE_PATH=/path/to/cache
Error: FileNotFoundError: [Errno 2] No such file or directory: 'pricingConfig.json'
- Context: Option pricing looks for config in repo root
- Fix: Ensure
pricingConfig.jsonexists or specify config explicitly - Default Values: Config is optional; defaults will be used if missing
Error: FileNotFoundError when accessing logs or cache directories
- Fix: Use absolute paths or ensure working directory is repo root
- Check:
os.getcwd()should return.../QuantTools - Workaround: Set
WORK_DIRenvironment variable explicitly
Unit Tests:
# Run all tests (if pytest configured)
pytest
# Run specific test file
python -m pytest tests/test_portfolio.py -vModule Tests (in module_test/):
# Test specific module
python module_test/test_backtest.py
# Test with verbose logging
python module_test/test_backtest.py --log-level DEBUGNotebook Tests:
- Open notebooks in
EventDriven/notebooks/orEventDriven/demos/ - Run cells sequentially
- Check for import errors or missing data files
Slow Backtests:
- Enable caching for market data
- Reduce option chain search space (fewer strikes, expirations)
- Use
slippage_enabled=Falsefor faster runs (testing only) - Profile with
cProfile(seeEventDriven/demos/demoRun.pyfor example)
Memory Issues:
- Clear cache periodically:
cache.clear() - Reduce backtest date range
- Limit number of symbols
- Use
delto remove large objects after use
Too Much Log Output:
from EventDriven.configs.core import BacktesterConfig
config = BacktesterConfig(logger_override_level='WARNING') # Reduce verbosityLogs Not Appearing:
- Check console output first (stdout/stderr)
- Verify
logs/directory exists - Check log level:
logger.setLevel(logging.DEBUG)
QuantTools uses Ruff for linting and formatting (configured in ruff.toml):
# Check for issues
ruff check .
# Auto-fix fixable issues
ruff check --fix .
# Format code
ruff format .Key Rules:
- Line length: 120 characters
- Import organization: Auto-sorted
- Ignored rules:
E501(line length, handled by formatter),I001(import order),E722(bare except),B009(getattr with constant)
- Type hints required for all public functions and methods
- Docstrings required for classes and public methods (Google or NumPy style)
- Use
from typing importfor compatibility with Python 3.10+
Example:
from typing import Optional
import pandas as pd
def calculate_returns(prices: pd.Series, periods: int = 252) -> pd.Series:
"""
Calculate period returns from price series.
Args:
prices: Time series of prices
periods: Number of periods for annualization (default: 252 for daily)
Returns:
Series of period returns
Raises:
ValueError: If prices series is empty
"""
if prices.empty:
raise ValueError("Price series cannot be empty")
return prices.pct_change(periods=periods)- Create Feature Branch:
git checkout -b feature/your-feature-nameorgit checkout -b YOURNAME-MMDD-feature-description - Make Changes: Follow code style, add tests
- Commit: Use descriptive commit messages
- Test: Run relevant tests before pushing
- Push:
git push origin your-branch-name - Pull Request: Submit PR to
mainwith description
Branch Naming Examples:
feature/add-iron-condor-strategyfix/portfolio-cash-calculationCHIDI-JAN04-DATAMANAGER-REBUILD(for larger initiatives)
- Add tests for new features in
module_test/ortests/ - Test edge cases (empty data, missing fields, extreme values)
- Use fixtures for common test data
- Document test assumptions in docstrings
MIT License - see LICENSE file for details
- Chidi - Core architecture, event-driven engine, options pricing
- Zino - Risk management, portfolio systems, data infrastructure
- Event-driven architecture inspired by QuantStart tutorials
- Option pricing models based on industry-standard formulas (Black-Scholes-Merton, etc.)
- Built with support from the quantitative finance community
Questions or Issues? Check existing issues on GitHub or open a new one with a minimal reproducible example.