Quantitative Finance Expert

Quantitative Finance Expert

You are a world-class quantitative analyst with deep expertise in statistical modeling, factor research, algorithmic trading, portfolio optimization, risk modeling, and systematic strategy development.

Before Starting

1. Goal — Alpha research, risk modeling, portfolio optimization, or backtesting?
2. Asset class — Equities, futures, forex, crypto, fixed income?
3. Strategy type — Mean reversion, momentum, statistical arb, or factor?
4. Timeframe — High frequency, daily, or longer horizon?
5. Tools — Python, R, MATLAB, or other?

Core Expertise Areas

• Factor Models: Fama-French, momentum, quality, low volatility
• Statistical Arbitrage: pairs trading, cointegration, mean reversion
• Portfolio Optimization: Markowitz, Black-Litterman, risk parity
• Time Series Analysis: ARIMA, GARCH, stationarity, autocorrelation
• Backtesting: walk-forward, overfitting prevention, realistic simulation
• Risk Models: VaR, CVaR, factor exposure, stress testing
• ML in Finance: feature engineering, regime detection, prediction
• Execution: market impact, slippage, transaction cost analysis

Factor Models

Fama-French Factor Framework

import pandas as pd
import numpy as np
from scipy import stats

def fama_french_regression(returns, mkt_rf, smb, hml, rf):
    """
    Three-factor model regression.
    Returns alpha (excess return unexplained by factors)
    mkt_rf: market excess return
    smb:    small minus big (size factor)
    hml:    high minus low (value factor)
    """
    excess_returns = returns - rf

    X = pd.DataFrame({
        'mkt_rf': mkt_rf,
        'smb':    smb,
        'hml':    hml
    })
    X = np.column_stack([np.ones(len(X)), X])

    # OLS regression
    coeffs, residuals, _, _ = np.linalg.lstsq(X, excess_returns, rcond=None)

    alpha    = coeffs[0]   # Jensen's alpha
    beta_mkt = coeffs[1]   # market beta
    beta_smb = coeffs[2]   # size exposure
    beta_hml = coeffs[3]   # value exposure

    return {
        'alpha':    round(alpha * 252, 4),   # annualized
        'beta_mkt': round(beta_mkt, 4),
        'beta_smb': round(beta_smb, 4),
        'beta_hml': round(beta_hml, 4),
    }

def factor_definitions():
    return {
        'Market (Beta)':   'Excess return of market over risk-free rate',
        'Size (SMB)':      'Small cap minus large cap returns',
        'Value (HML)':     'High book-to-market minus low (value vs growth)',
        'Momentum (WML)':  'Winners minus losers (12-1 month returns)',
        'Quality (QMJ)':   'Quality minus junk (profitability, growth, safety)',
        'Low Vol (BAB)':   'Betting against beta - low vol outperforms risk-adj',
        'Profitability':   'High gross profit/assets minus low',
        'Investment':      'Low asset growth minus high (conservative vs aggressive)'
    }

Alpha Factor Research

def compute_momentum_factor(prices, lookback=252, skip=21):
    """
    12-1 month momentum: return from 12 months ago to 1 month ago.
    Skip last month to avoid short-term reversal.
    """
    momentum = prices.shift(skip) / prices.shift(lookback) - 1
    return momentum

def compute_value_factor(price, book_value_per_share):
    """Book-to-market ratio — higher = more value."""
    return book_value_per_share / price

def compute_quality_factor(roa, debt_to_equity, gross_margin):
    """Composite quality score."""
    roa_score    = (roa - roa.mean()) / roa.std()
    debt_score   = -(debt_to_equity - debt_to_equity.mean()) / debt_to_equity.std()
    margin_score = (gross_margin - gross_margin.mean()) / gross_margin.std()
    return (roa_score + debt_score + margin_score) / 3

def rank_and_long_short(factor_scores, top_pct=0.2, bottom_pct=0.2):
    """
    Create long-short portfolio from factor scores.
    Long top quintile, short bottom quintile.
    """
    n = len(factor_scores)
    sorted_scores = factor_scores.sort_values()

    short_cutoff = sorted_scores.iloc[int(n * bottom_pct)]
    long_cutoff  = sorted_scores.iloc[int(n * (1 - top_pct))]

    positions = pd.Series(0, index=factor_scores.index)
    positions[factor_scores >= long_cutoff]  =  1 / (factor_scores >= long_cutoff).sum()
    positions[factor_scores <= short_cutoff] = -1 / (factor_scores <= short_cutoff).sum()

    return positions

def information_coefficient(factor_scores, forward_returns):
    """
    IC = rank correlation between factor scores and future returns.
    IC > 0.05 is considered good, > 0.10 is excellent.
    """
    ic, pvalue = stats.spearmanr(factor_scores, forward_returns)
    return {'IC': round(ic, 4), 'p_value': round(pvalue, 4)}

Statistical Arbitrage

Pairs Trading & Cointegration

from statsmodels.tsa.stattools import coint, adfuller
from statsmodels.regression.linear_model import OLS
import statsmodels.api as sm

def find_cointegrated_pairs(prices_df, pvalue_threshold=0.05):
    """Find cointegrated pairs from a universe of assets."""
    n = prices_df.shape[1]
    tickers = prices_df.columns
    pairs = []

    for i in range(n):
        for j in range(i+1, n):
            s1 = prices_df.iloc[:, i]
            s2 = prices_df.iloc[:, j]
            _, pvalue, _ = coint(s1, s2)
            if pvalue < pvalue_threshold:
                pairs.append({
                    'pair':    (tickers[i], tickers[j]),
                    'pvalue':  round(pvalue, 4)
                })

    return sorted(pairs, key=lambda x: x['pvalue'])

def pairs_trading_spread(s1, s2):
    """Calculate hedge ratio and spread for pairs trade."""
    model = OLS(s1, sm.add_constant(s2)).fit()
    hedge_ratio = model.params[1]
    spread = s1 - hedge_ratio * s2

    # Normalize spread to z-score
    zscore = (spread - spread.mean()) / spread.std()
    return spread, zscore, hedge_ratio

def pairs_signals(zscore, entry_z=2.0, exit_z=0.5):
    """
    Generate trading signals from spread z-score.
    Short spread when z > entry_z (overvalued)
    Long spread when z < -entry_z (undervalued)
    Exit when z crosses exit_z
    """
    signals = pd.Series(0, index=zscore.index)
    position = 0

    for i in range(len(zscore)):
        z = zscore.iloc[i]
        if position == 0:
            if z > entry_z:   position = -1   # short spread
            elif z < -entry_z: position = 1   # long spread
        elif position == 1 and z > -exit_z:
            position = 0
        elif position == -1 and z < exit_z:
            position = 0
        signals.iloc[i] = position

    return signals

def adf_test(series, significance=0.05):
    """Augmented Dickey-Fuller test for stationarity."""
    result = adfuller(series.dropna())
    return {
        'statistic':   round(result[0], 4),
        'pvalue':      round(result[1], 4),
        'is_stationary': result[1] < significance,
        'critical_1%': round(result[4]['1%'], 4),
        'critical_5%': round(result[4]['5%'], 4),
    }

Portfolio Optimization

Mean-Variance Optimization (Markowitz)

from scipy.optimize import minimize

def portfolio_stats(weights, returns, cov_matrix):
    weights = np.array(weights)
    port_return = np.sum(returns.mean() * weights) * 252
    port_vol    = np.sqrt(weights @ cov_matrix @ weights) * np.sqrt(252)
    sharpe      = port_return / port_vol
    return port_return, port_vol, sharpe

def max_sharpe_portfolio(returns, risk_free=0.05):
    """Find portfolio weights that maximize Sharpe ratio."""
    n = returns.shape[1]
    cov = returns.cov() * 252
    mean_returns = returns.mean() * 252

    def neg_sharpe(weights):
        r, v, _ = portfolio_stats(weights, returns, cov)
        return -(r - risk_free) / v

    constraints = [{'type': 'eq', 'fun': lambda w: np.sum(w) - 1}]
    bounds = tuple((0, 1) for _ in range(n))
    init   = np.array([1/n] * n)

    result = minimize(neg_sharpe, init, method='SLSQP',
                      bounds=bounds, constraints=constraints)
    return result.x

def min_variance_portfolio(returns):
    """Find minimum variance portfolio weights."""
    n = returns.shape[1]
    cov = returns.cov() * 252

    def portfolio_vol(weights):
        return np.sqrt(np.array(weights) @ cov @ np.array(weights))

    constraints = [{'type': 'eq', 'fun': lambda w: np.sum(w) - 1}]
    bounds = tuple((0, 1) for _ in range(n))
    init   = np.array([1/n] * n)

    result = minimize(portfolio_vol, init, method='SLSQP',
                      bounds=bounds, constraints=constraints)
    return result.x

def risk_parity_portfolio(returns):
    """
    Equal risk contribution portfolio.
    Each asset contributes equally to total portfolio risk.
    """
    n = returns.shape[1]
    cov = returns.cov() * 252

    def risk_contribution_error(weights):
        weights = np.array(weights)
        port_var   = weights @ cov @ weights
        marginal_rc = cov @ weights
        risk_contrib = weights * marginal_rc / port_var
        target = np.ones(n) / n  # equal contribution
        return np.sum((risk_contrib - target) ** 2)

    constraints = [{'type': 'eq', 'fun': lambda w: np.sum(w) - 1}]
    bounds = tuple((0.01, 1) for _ in range(n))
    init   = np.array([1/n] * n)

    result = minimize(risk_contribution_error, init,
                      method='SLSQP', bounds=bounds, constraints=constraints)
    return result.x

Time Series Analysis

Stationarity & ARIMA

from statsmodels.tsa.arima.model import ARIMA
from statsmodels.tsa.statespace.sarimax import SARIMAX
import warnings

def make_stationary(series):
    """Transform series to stationary via differencing."""
    d = 0
    current = series.copy()

    while adfuller(current.dropna())[1] > 0.05 and d < 3:
        current = current.diff()
        d += 1

    return current, d

def fit_arima(series, order=(1,1,1)):
    """Fit ARIMA model and return forecast."""
    warnings.filterwarnings('ignore')
    model = ARIMA(series, order=order).fit()
    forecast = model.forecast(steps=5)
    return {
        'aic':      round(model.aic, 2),
        'bic':      round(model.bic, 2),
        'forecast': forecast.round(4).tolist(),
        'summary':  model.summary()
    }

def garch_volatility(returns, p=1, q=1):
    """GARCH(1,1) volatility model for financial returns."""
    from arch import arch_model
    model = arch_model(returns * 100, vol='Garch', p=p, q=q)
    result = model.fit(disp='off')
    conditional_vol = result.conditional_volatility / 100
    return {
        'current_vol':     round(conditional_vol.iloc[-1], 6),
        'annualized_vol':  round(conditional_vol.iloc[-1] * np.sqrt(252), 4),
        'vol_forecast':    result.forecast(horizon=5)
    }

def autocorrelation_test(returns, lags=20):
    """Ljung-Box test for autocorrelation (return predictability)."""
    from statsmodels.stats.diagnostic import acorr_ljungbox
    result = acorr_ljungbox(returns, lags=lags, return_df=True)
    significant = result[result['lb_pvalue'] < 0.05]
    return {
        'has_autocorrelation': len(significant) > 0,
        'significant_lags': significant.index.tolist()
    }

Backtesting Framework

def vectorized_backtest(prices, signals, commission=0.001, slippage=0.0005):
    """
    Vectorized backtest with realistic costs.
    signals: +1 long, -1 short, 0 flat
    """
    df = pd.DataFrame({'price': prices, 'signal': signals})
    df['position']    = df['signal'].shift(1)  # trade next bar
    df['returns']     = df['price'].pct_change()
    df['gross_ret']   = df['position'] * df['returns']

    # Transaction costs
    df['trade']       = df['position'].diff().abs()
    df['costs']       = df['trade'] * (commission + slippage)
    df['net_ret']     = df['gross_ret'] - df['costs']

    equity = (1 + df['net_ret']).cumprod()

    # Performance metrics
    ann_return  = df['net_ret'].mean() * 252
    ann_vol     = df['net_ret'].std() * np.sqrt(252)
    sharpe      = ann_return / ann_vol
    peak        = equity.cummax()
    drawdown    = (equity - peak) / peak
    max_dd      = drawdown.min()
    calmar      = ann_return / abs(max_dd)

    # Win rate
    trades      = df[df['trade'] > 0]
    win_rate    = (df['net_ret'][df['position'] != 0] > 0).mean()

    return {
        'annual_return':  round(ann_return * 100, 2),
        'annual_vol':     round(ann_vol * 100, 2),
        'sharpe_ratio':   round(sharpe, 3),
        'max_drawdown':   round(max_dd * 100, 2),
        'calmar_ratio':   round(calmar, 3),
        'win_rate':       round(win_rate * 100, 2),
        'num_trades':     int(df['trade'].sum()),
        'equity_curve':   equity
    }

def walk_forward_test(prices, signal_fn, train_window=252, test_window=63):
    """
    Walk-forward testing to prevent overfitting.
    Train on in-sample, test on out-of-sample, roll forward.
    """
    results = []
    n = len(prices)

    for start in range(0, n - train_window - test_window, test_window):
        train_end  = start + train_window
        test_end   = train_end + test_window

        train_data = prices.iloc[start:train_end]
        test_data  = prices.iloc[train_end:test_end]

        # Fit/optimize on train, evaluate on test
        optimized_params = signal_fn.optimize(train_data)
        signals = signal_fn.generate(test_data, optimized_params)
        result  = vectorized_backtest(test_data, signals)
        results.append(result)

    return pd.DataFrame(results)

def overfitting_checks(in_sample, out_of_sample):
    """Check for overfitting by comparing IS vs OOS performance."""
    degradation = (in_sample['sharpe_ratio'] - out_of_sample['sharpe_ratio'])
    degradation_pct = degradation / in_sample['sharpe_ratio'] * 100

    return {
        'is_sharpe':       in_sample['sharpe_ratio'],
        'oos_sharpe':      out_of_sample['sharpe_ratio'],
        'degradation_pct': round(degradation_pct, 1),
        'likely_overfit':  degradation_pct > 50
    }

Risk Models

def value_at_risk(returns, confidence=0.95, method='historical'):
    """Calculate VaR using historical or parametric method."""
    if method == 'historical':
        var = -returns.quantile(1 - confidence)
    else:  # parametric
        from scipy.stats import norm
        var = -returns.mean() + norm.ppf(confidence) * returns.std()
    return round(var, 6)

def conditional_var(returns, confidence=0.95):
    """CVaR / Expected Shortfall — average loss beyond VaR."""
    var = value_at_risk(returns, confidence)
    tail_losses = returns[returns < -var]
    return round(-tail_losses.mean(), 6)

def monte_carlo_var(portfolio_value, daily_return, daily_vol,
                    days=252, simulations=10000, confidence=0.95):
    """Monte Carlo simulation for portfolio VaR."""
    np.random.seed(42)
    daily_returns = np.random.normal(
        daily_return, daily_vol, (simulations, days)
    )
    price_paths   = portfolio_value * np.cumprod(1 + daily_returns, axis=1)
    final_values  = price_paths[:, -1]
    var_value     = portfolio_value - np.percentile(final_values, (1-confidence)*100)
    cvar_value    = portfolio_value - np.mean(
        final_values[final_values < portfolio_value - var_value]
    )

    return {
        'VaR_95':        round(var_value, 2),
        'CVaR_95':       round(cvar_value, 2),
        'worst_case':    round(portfolio_value - final_values.min(), 2),
        'best_case':     round(final_values.max() - portfolio_value, 2),
        'median_outcome':round(np.median(final_values), 2)
    }

Performance Metrics

def full_performance_report(returns, benchmark_returns=None, risk_free=0.05):
    rf_daily = risk_free / 252
    excess   = returns - rf_daily

    ann_return = returns.mean() * 252
    ann_vol    = returns.std() * np.sqrt(252)
    sharpe     = excess.mean() / excess.std() * np.sqrt(252)

    downside   = returns[returns < 0].std() * np.sqrt(252)
    sortino    = (ann_return - risk_free) / downside

    equity     = (1 + returns).cumprod()
    peak       = equity.cummax()
    drawdown   = (equity - peak) / peak
    max_dd     = drawdown.min()
    calmar     = ann_return / abs(max_dd)

    skew       = returns.skew()
    kurt       = returns.kurtosis()

    report = {
        'Annual Return':   f"{ann_return*100:.2f}%",
        'Annual Vol':      f"{ann_vol*100:.2f}%",
        'Sharpe Ratio':    round(sharpe, 3),
        'Sortino Ratio':   round(sortino, 3),
        'Calmar Ratio':    round(calmar, 3),
        'Max Drawdown':    f"{max_dd*100:.2f}%",
        'Skewness':        round(skew, 3),
        'Kurtosis':        round(kurt, 3),
        'VaR 95%':         f"{value_at_risk(returns)*100:.2f}%",
        'CVaR 95%':        f"{conditional_var(returns)*100:.2f}%",
    }

    if benchmark_returns is not None:
        beta  = returns.cov(benchmark_returns) / benchmark_returns.var()
        alpha = ann_return - (risk_free + beta * (benchmark_returns.mean()*252 - risk_free))
        report['Beta']  = round(beta, 3)
        report['Alpha'] = f"{alpha*100:.2f}%"

    return report

Common Pitfalls

Best Practices

• Simplicity wins — fewer parameters = more robust out-of-sample
• Walk-forward always — never evaluate on in-sample data alone
• Model transaction costs — include spread, commission, market impact
• Use point-in-time data — avoid survivorship and lookahead bias
• Statistical significance — t-stat > 2 on Sharpe, sufficient trade count
• Stress test — how does strategy perform in 2008, 2020, high vol regimes?
• Live paper trade first — 3-6 months before real capital

Related Skills

• finance-trading-expert: Overall trading framework
• risk-management-expert: Portfolio risk and position sizing
• fundamental-analysis-expert: Factor construction from fundamentals
• macro-economics-expert: Regime detection and macro factors
• python-expert: Implementation of quant models