791 lines
33 KiB
Plaintext
791 lines
33 KiB
Plaintext
|
import pandas as pd
|
||
|
|
import numpy as np
|
||
|
|
import matplotlib.pyplot as plt
|
||
|
|
import seaborn as sns
|
||
|
|
from typing import Tuple, Dict, List, Optional, Union
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from scipy import stats
|
||
|
|
import warnings
|
||
|
|
warnings.filterwarnings('ignore')
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class DrawdownMetrics:
|
||
|
|
"""Container for drawdown analysis results"""
|
||
|
|
avg_drawdown: float
|
||
|
|
max_drawdown: float
|
||
|
|
recovery_count: int
|
||
|
|
recovery_probability: float
|
||
|
|
avg_recovery_time: float
|
||
|
|
confidence_interval: Tuple[float, float]
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class OrderBookSnapshot:
|
||
|
|
"""Simulated order book data"""
|
||
|
|
bid_depth: float
|
||
|
|
ask_depth: float
|
||
|
|
spread: float
|
||
|
|
imbalance: float
|
||
|
|
|
||
|
|
class ConditionalDrawdownAnalyzer:
|
||
|
|
"""
|
||
|
|
Complete framework for conditional drawdown analysis with all enhancements:
|
||
|
|
- Recursive self-consistent drawdowns
|
||
|
|
- Volatility adjustment
|
||
|
|
- Order book integration
|
||
|
|
- Backtesting capabilities
|
||
|
|
- Regime detection
|
||
|
|
- Entry optimization
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, confidence_level: float = 0.95):
|
||
|
|
self.confidence_level = confidence_level
|
||
|
|
self.results_cache = {}
|
||
|
|
|
||
|
|
def basic_conditional_drawdown(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_price: float) -> DrawdownMetrics:
|
||
|
|
"""
|
||
|
|
Basic conditional drawdown: track lows below entry until recovery above entry
|
||
|
|
"""
|
||
|
|
below_entry = price < entry_price
|
||
|
|
lows = []
|
||
|
|
recovery_times = []
|
||
|
|
i = 0
|
||
|
|
n = len(price)
|
||
|
|
|
||
|
|
while i < n:
|
||
|
|
if below_entry.iloc[i]:
|
||
|
|
start_idx = i
|
||
|
|
min_price = price.iloc[i]
|
||
|
|
|
||
|
|
# Track the drawdown period
|
||
|
|
while i < n and price.iloc[i] < entry_price:
|
||
|
|
min_price = min(min_price, price.iloc[i])
|
||
|
|
i += 1
|
||
|
|
|
||
|
|
# Check if we recovered (reached end or price >= entry)
|
||
|
|
if i < n and price.iloc[i] >= entry_price:
|
||
|
|
lows.append(min_price)
|
||
|
|
recovery_times.append(i - start_idx)
|
||
|
|
elif i == n and len(lows) > 0:
|
||
|
|
# End of series - only count if we had previous recoveries
|
||
|
|
lows.append(min_price)
|
||
|
|
recovery_times.append(i - start_idx)
|
||
|
|
else:
|
||
|
|
i += 1
|
||
|
|
|
||
|
|
if not lows:
|
||
|
|
return DrawdownMetrics(0, 0, 0, 0, 0, (0, 0))
|
||
|
|
|
||
|
|
drawdowns = [entry_price - low for low in lows]
|
||
|
|
avg_dd = np.mean(drawdowns)
|
||
|
|
max_dd = np.max(drawdowns)
|
||
|
|
|
||
|
|
# Confidence interval
|
||
|
|
if len(drawdowns) > 1:
|
||
|
|
ci = stats.t.interval(self.confidence_level, len(drawdowns)-1,
|
||
|
|
loc=avg_dd, scale=stats.sem(drawdowns))
|
||
|
|
else:
|
||
|
|
ci = (avg_dd, avg_dd)
|
||
|
|
|
||
|
|
return DrawdownMetrics(
|
||
|
|
avg_drawdown=avg_dd,
|
||
|
|
max_drawdown=max_dd,
|
||
|
|
recovery_count=len(lows),
|
||
|
|
recovery_probability=len(lows) / max(1, np.sum(below_entry)),
|
||
|
|
avg_recovery_time=np.mean(recovery_times) if recovery_times else 0,
|
||
|
|
confidence_interval=ci
|
||
|
|
)
|
||
|
|
|
||
|
|
def recursive_conditional_drawdown(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_price: float,
|
||
|
|
tol: float = 1e-5,
|
||
|
|
max_iter: int = 50) -> DrawdownMetrics:
|
||
|
|
"""
|
||
|
|
Recursive conditional drawdown where recovery must exceed entry + avg_drawdown
|
||
|
|
"""
|
||
|
|
avg_dd = 0.0
|
||
|
|
|
||
|
|
for iteration in range(max_iter):
|
||
|
|
lows = []
|
||
|
|
recovery_times = []
|
||
|
|
i = 0
|
||
|
|
n = len(price)
|
||
|
|
recovery_threshold = entry_price + avg_dd
|
||
|
|
|
||
|
|
while i < n:
|
||
|
|
if price.iloc[i] < entry_price:
|
||
|
|
start_idx = i
|
||
|
|
min_price = price.iloc[i]
|
||
|
|
|
||
|
|
# Track drawdown until sufficient recovery
|
||
|
|
while i < n and price.iloc[i] < recovery_threshold:
|
||
|
|
min_price = min(min_price, price.iloc[i])
|
||
|
|
i += 1
|
||
|
|
|
||
|
|
# Only count if we actually recovered above threshold
|
||
|
|
if i < n and price.iloc[i] >= recovery_threshold:
|
||
|
|
lows.append(min_price)
|
||
|
|
recovery_times.append(i - start_idx)
|
||
|
|
else:
|
||
|
|
i += 1
|
||
|
|
|
||
|
|
if lows:
|
||
|
|
new_avg_dd = np.mean([entry_price - low for low in lows])
|
||
|
|
else:
|
||
|
|
new_avg_dd = 0.0
|
||
|
|
|
||
|
|
if abs(new_avg_dd - avg_dd) < tol:
|
||
|
|
break
|
||
|
|
avg_dd = new_avg_dd
|
||
|
|
|
||
|
|
if not lows:
|
||
|
|
return DrawdownMetrics(0, 0, 0, 0, 0, (0, 0))
|
||
|
|
|
||
|
|
drawdowns = [entry_price - low for low in lows]
|
||
|
|
max_dd = np.max(drawdowns)
|
||
|
|
|
||
|
|
# Confidence interval
|
||
|
|
if len(drawdowns) > 1:
|
||
|
|
ci = stats.t.interval(self.confidence_level, len(drawdowns)-1,
|
||
|
|
loc=avg_dd, scale=stats.sem(drawdowns))
|
||
|
|
else:
|
||
|
|
ci = (avg_dd, avg_dd)
|
||
|
|
|
||
|
|
return DrawdownMetrics(
|
||
|
|
avg_drawdown=avg_dd,
|
||
|
|
max_drawdown=max_dd,
|
||
|
|
recovery_count=len(lows),
|
||
|
|
recovery_probability=len(lows) / max(1, len(price)),
|
||
|
|
avg_recovery_time=np.mean(recovery_times) if recovery_times else 0,
|
||
|
|
confidence_interval=ci
|
||
|
|
)
|
||
|
|
|
||
|
|
def calculate_volatility(self,
|
||
|
|
price: pd.Series,
|
||
|
|
method: str = 'atr',
|
||
|
|
window: int = 14) -> pd.Series:
|
||
|
|
"""
|
||
|
|
Calculate volatility using various methods
|
||
|
|
"""
|
||
|
|
if method == 'atr':
|
||
|
|
# Approximate ATR using price changes
|
||
|
|
returns = price.pct_change().abs()
|
||
|
|
return returns.rolling(window).mean() * price
|
||
|
|
elif method == 'std':
|
||
|
|
returns = price.pct_change()
|
||
|
|
return returns.rolling(window).std() * price
|
||
|
|
elif method == 'ewm':
|
||
|
|
returns = price.pct_change()
|
||
|
|
return returns.ewm(span=window).std() * price
|
||
|
|
else:
|
||
|
|
raise ValueError(f"Unknown volatility method: {method}")
|
||
|
|
|
||
|
|
def volatility_adjusted_drawdown(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_price: float,
|
||
|
|
vol_method: str = 'atr',
|
||
|
|
vol_window: int = 14,
|
||
|
|
recursive: bool = True) -> Dict:
|
||
|
|
"""
|
||
|
|
Calculate volatility-adjusted conditional drawdowns
|
||
|
|
"""
|
||
|
|
# Get base drawdown
|
||
|
|
if recursive:
|
||
|
|
base_metrics = self.recursive_conditional_drawdown(price, entry_price)
|
||
|
|
else:
|
||
|
|
base_metrics = self.basic_conditional_drawdown(price, entry_price)
|
||
|
|
|
||
|
|
# Calculate volatility at entry point
|
||
|
|
volatility = self.calculate_volatility(price, vol_method, vol_window)
|
||
|
|
entry_idx = (price - entry_price).abs().idxmin()
|
||
|
|
entry_vol = volatility.loc[entry_idx] if entry_idx in volatility.index else volatility.mean()
|
||
|
|
|
||
|
|
# Volatility adjustment coefficient
|
||
|
|
avg_vol = volatility.mean()
|
||
|
|
vol_coeff = entry_vol / avg_vol if avg_vol > 0 else 1.0
|
||
|
|
|
||
|
|
adjusted_metrics = DrawdownMetrics(
|
||
|
|
avg_drawdown=base_metrics.avg_drawdown / vol_coeff,
|
||
|
|
max_drawdown=base_metrics.max_drawdown / vol_coeff,
|
||
|
|
recovery_count=base_metrics.recovery_count,
|
||
|
|
recovery_probability=base_metrics.recovery_probability,
|
||
|
|
avg_recovery_time=base_metrics.avg_recovery_time,
|
||
|
|
confidence_interval=(base_metrics.confidence_interval[0] / vol_coeff,
|
||
|
|
base_metrics.confidence_interval[1] / vol_coeff)
|
||
|
|
)
|
||
|
|
|
||
|
|
return {
|
||
|
|
'base_metrics': base_metrics,
|
||
|
|
'adjusted_metrics': adjusted_metrics,
|
||
|
|
'volatility_coefficient': vol_coeff,
|
||
|
|
'entry_volatility': entry_vol,
|
||
|
|
'average_volatility': avg_vol
|
||
|
|
}
|
||
|
|
|
||
|
|
def simulate_order_book(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_price: float) -> OrderBookSnapshot:
|
||
|
|
"""
|
||
|
|
Simulate order book metrics based on price action
|
||
|
|
"""
|
||
|
|
# Simple simulation based on recent volatility and volume proxy
|
||
|
|
recent_vol = price.pct_change().tail(20).std()
|
||
|
|
distance_to_entry = abs(price.iloc[-1] - entry_price) / entry_price
|
||
|
|
|
||
|
|
# Simulate depth (higher vol = lower depth)
|
||
|
|
base_depth = 10000
|
||
|
|
bid_depth = base_depth * (1 - recent_vol * 10)
|
||
|
|
ask_depth = base_depth * (1 - recent_vol * 10)
|
||
|
|
|
||
|
|
# Simulate spread (higher vol = wider spread)
|
||
|
|
spread = entry_price * (0.001 + recent_vol * 5)
|
||
|
|
|
||
|
|
# Simulate imbalance
|
||
|
|
imbalance = 0.5 + np.random.normal(0, 0.1) # Random around balanced
|
||
|
|
|
||
|
|
return OrderBookSnapshot(
|
||
|
|
bid_depth=max(1000, bid_depth),
|
||
|
|
ask_depth=max(1000, ask_depth),
|
||
|
|
spread=spread,
|
||
|
|
imbalance=max(0.1, min(0.9, imbalance))
|
||
|
|
)
|
||
|
|
|
||
|
|
def order_book_adjusted_drawdown(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_price: float,
|
||
|
|
order_book: Optional[OrderBookSnapshot] = None) -> Dict:
|
||
|
|
"""
|
||
|
|
Adjust drawdown based on order book liquidity
|
||
|
|
"""
|
||
|
|
base_result = self.volatility_adjusted_drawdown(price, entry_price)
|
||
|
|
|
||
|
|
if order_book is None:
|
||
|
|
order_book = self.simulate_order_book(price, entry_price)
|
||
|
|
|
||
|
|
# Liquidity adjustment factor
|
||
|
|
min_depth = min(order_book.bid_depth, order_book.ask_depth)
|
||
|
|
depth_factor = 1.0 + (10000 - min_depth) / 50000 # Higher adjustment for thin books
|
||
|
|
spread_factor = 1.0 + order_book.spread / entry_price * 100 # Spread penalty
|
||
|
|
imbalance_factor = 1.0 + abs(0.5 - order_book.imbalance) * 0.5 # Imbalance penalty
|
||
|
|
|
||
|
|
liquidity_adjustment = depth_factor * spread_factor * imbalance_factor
|
||
|
|
|
||
|
|
adjusted_metrics = base_result['adjusted_metrics']
|
||
|
|
liquidity_adjusted = DrawdownMetrics(
|
||
|
|
avg_drawdown=adjusted_metrics.avg_drawdown * liquidity_adjustment,
|
||
|
|
max_drawdown=adjusted_metrics.max_drawdown * liquidity_adjustment,
|
||
|
|
recovery_count=adjusted_metrics.recovery_count,
|
||
|
|
recovery_probability=adjusted_metrics.recovery_probability / liquidity_adjustment,
|
||
|
|
avg_recovery_time=adjusted_metrics.avg_recovery_time * liquidity_adjustment,
|
||
|
|
confidence_interval=(adjusted_metrics.confidence_interval[0] * liquidity_adjustment,
|
||
|
|
adjusted_metrics.confidence_interval[1] * liquidity_adjustment)
|
||
|
|
)
|
||
|
|
|
||
|
|
return {
|
||
|
|
**base_result,
|
||
|
|
'liquidity_adjusted_metrics': liquidity_adjusted,
|
||
|
|
'order_book': order_book,
|
||
|
|
'liquidity_adjustment_factor': liquidity_adjustment
|
||
|
|
}
|
||
|
|
|
||
|
|
def create_drawdown_surface(self,
|
||
|
|
price: pd.Series,
|
||
|
|
price_range: Optional[Tuple[float, float]] = None,
|
||
|
|
resolution: float = 0.005, # 0.5% steps
|
||
|
|
recursive: bool = True) -> pd.DataFrame:
|
||
|
|
"""
|
||
|
|
Create drawdown surface across all potential entry prices
|
||
|
|
"""
|
||
|
|
if price_range is None:
|
||
|
|
price_min, price_max = price.min() * 0.95, price.max() * 1.05
|
||
|
|
else:
|
||
|
|
price_min, price_max = price_range
|
||
|
|
|
||
|
|
# Create entry price grid
|
||
|
|
num_points = int((price_max - price_min) / (price_min * resolution))
|
||
|
|
entry_prices = np.linspace(price_min, price_max, min(num_points, 200)) # Limit for performance
|
||
|
|
|
||
|
|
results = []
|
||
|
|
for entry_price in entry_prices:
|
||
|
|
try:
|
||
|
|
if recursive:
|
||
|
|
metrics = self.recursive_conditional_drawdown(price, entry_price)
|
||
|
|
else:
|
||
|
|
metrics = self.basic_conditional_drawdown(price, entry_price)
|
||
|
|
|
||
|
|
results.append({
|
||
|
|
'entry_price': entry_price,
|
||
|
|
'avg_drawdown': metrics.avg_drawdown,
|
||
|
|
'max_drawdown': metrics.max_drawdown,
|
||
|
|
'recovery_count': metrics.recovery_count,
|
||
|
|
'recovery_probability': metrics.recovery_probability,
|
||
|
|
'avg_recovery_time': metrics.avg_recovery_time,
|
||
|
|
'ci_lower': metrics.confidence_interval[0],
|
||
|
|
'ci_upper': metrics.confidence_interval[1]
|
||
|
|
})
|
||
|
|
except Exception as e:
|
||
|
|
# Skip problematic entry prices
|
||
|
|
continue
|
||
|
|
|
||
|
|
return pd.DataFrame(results)
|
||
|
|
|
||
|
|
def detect_abnormal_regime(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_price: float,
|
||
|
|
current_drawdown: float,
|
||
|
|
threshold_multiplier: float = 2.0) -> Dict:
|
||
|
|
"""
|
||
|
|
Detect if current drawdown indicates abnormal market regime
|
||
|
|
"""
|
||
|
|
metrics = self.recursive_conditional_drawdown(price, entry_price)
|
||
|
|
|
||
|
|
# Check if current drawdown exceeds historical norms
|
||
|
|
expected_dd = metrics.avg_drawdown
|
||
|
|
ci_upper = metrics.confidence_interval[1]
|
||
|
|
|
||
|
|
is_abnormal = current_drawdown > (expected_dd * threshold_multiplier)
|
||
|
|
is_extreme = current_drawdown > ci_upper
|
||
|
|
|
||
|
|
severity_score = current_drawdown / max(expected_dd, 0.001) # Avoid division by zero
|
||
|
|
|
||
|
|
return {
|
||
|
|
'is_abnormal_regime': is_abnormal,
|
||
|
|
'is_extreme_drawdown': is_extreme,
|
||
|
|
'severity_score': severity_score,
|
||
|
|
'expected_drawdown': expected_dd,
|
||
|
|
'current_drawdown': current_drawdown,
|
||
|
|
'confidence_upper': ci_upper,
|
||
|
|
'recommendation': self._get_regime_recommendation(severity_score, is_abnormal, is_extreme)
|
||
|
|
}
|
||
|
|
|
||
|
|
def _get_regime_recommendation(self, severity_score: float, is_abnormal: bool, is_extreme: bool) -> str:
|
||
|
|
"""Generate trading recommendations based on regime analysis"""
|
||
|
|
if is_extreme:
|
||
|
|
return "EXTREME: Consider immediate exit or significant position reduction"
|
||
|
|
elif is_abnormal:
|
||
|
|
return "ABNORMAL: Tighten stops, reduce position size, or hedge"
|
||
|
|
elif severity_score > 1.5:
|
||
|
|
return "ELEVATED: Monitor closely, consider partial profit-taking"
|
||
|
|
else:
|
||
|
|
return "NORMAL: Continue with current strategy"
|
||
|
|
|
||
|
|
def optimize_entries(self,
|
||
|
|
price: pd.Series,
|
||
|
|
reward_target: float = 0.02, # 2% target
|
||
|
|
max_entries: int = 10) -> pd.DataFrame:
|
||
|
|
"""
|
||
|
|
Find optimal entry points based on risk-reward ratio
|
||
|
|
"""
|
||
|
|
surface = self.create_drawdown_surface(price)
|
||
|
|
|
||
|
|
if surface.empty:
|
||
|
|
return pd.DataFrame()
|
||
|
|
|
||
|
|
# Calculate risk-reward ratios
|
||
|
|
surface['potential_reward'] = reward_target
|
||
|
|
surface['risk_reward_ratio'] = surface['potential_reward'] / surface['avg_drawdown'].replace(0, np.inf)
|
||
|
|
surface['score'] = (surface['risk_reward_ratio'] * surface['recovery_probability'] *
|
||
|
|
(1 / (surface['avg_recovery_time'] + 1)))
|
||
|
|
|
||
|
|
# Filter out low-probability entries
|
||
|
|
surface_filtered = surface[surface['recovery_probability'] > 0.3].copy()
|
||
|
|
|
||
|
|
if surface_filtered.empty:
|
||
|
|
return surface.nlargest(max_entries, 'score')
|
||
|
|
|
||
|
|
# Get top entries
|
||
|
|
optimal_entries = surface_filtered.nlargest(max_entries, 'score')
|
||
|
|
|
||
|
|
return optimal_entries[['entry_price', 'avg_drawdown', 'recovery_probability',
|
||
|
|
'risk_reward_ratio', 'score']].round(4)
|
||
|
|
|
||
|
|
def backtest_strategy(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_signals: pd.Series,
|
||
|
|
exit_method: str = 'adaptive_stop',
|
||
|
|
initial_capital: float = 100000) -> Dict:
|
||
|
|
"""
|
||
|
|
Backtest trading strategy using conditional drawdown insights
|
||
|
|
"""
|
||
|
|
trades = []
|
||
|
|
capital = initial_capital
|
||
|
|
position = 0
|
||
|
|
entry_price = 0
|
||
|
|
|
||
|
|
for i, (timestamp, signal) in enumerate(entry_signals.items()):
|
||
|
|
current_price = price.loc[timestamp]
|
||
|
|
|
||
|
|
if signal > 0 and position == 0: # Entry signal
|
||
|
|
# Calculate position size based on drawdown risk
|
||
|
|
metrics = self.recursive_conditional_drawdown(price.loc[:timestamp], current_price)
|
||
|
|
risk_amount = capital * 0.02 # Risk 2% per trade
|
||
|
|
|
||
|
|
if metrics.avg_drawdown > 0:
|
||
|
|
position_size = risk_amount / metrics.avg_drawdown
|
||
|
|
position = min(position_size, capital / current_price) # Can't buy more than we have
|
||
|
|
entry_price = current_price
|
||
|
|
|
||
|
|
elif position > 0: # Manage existing position
|
||
|
|
current_drawdown = max(0, entry_price - current_price)
|
||
|
|
|
||
|
|
# Check for exit conditions
|
||
|
|
should_exit = False
|
||
|
|
exit_reason = ""
|
||
|
|
|
||
|
|
if exit_method == 'adaptive_stop':
|
||
|
|
# Use conditional drawdown as adaptive stop
|
||
|
|
metrics = self.recursive_conditional_drawdown(price.loc[:timestamp], entry_price)
|
||
|
|
if current_drawdown > metrics.avg_drawdown * 1.5:
|
||
|
|
should_exit = True
|
||
|
|
exit_reason = "Adaptive stop loss"
|
||
|
|
|
||
|
|
elif exit_method == 'regime_detection':
|
||
|
|
regime_info = self.detect_abnormal_regime(price.loc[:timestamp], entry_price, current_drawdown)
|
||
|
|
if regime_info['is_abnormal_regime']:
|
||
|
|
should_exit = True
|
||
|
|
exit_reason = "Abnormal regime detected"
|
||
|
|
|
||
|
|
# Exit on signal or end of data
|
||
|
|
if should_exit or signal < 0 or i == len(entry_signals) - 1:
|
||
|
|
pnl = position * (current_price - entry_price)
|
||
|
|
capital += pnl
|
||
|
|
|
||
|
|
trades.append({
|
||
|
|
'entry_time': timestamp,
|
||
|
|
'entry_price': entry_price,
|
||
|
|
'exit_time': timestamp,
|
||
|
|
'exit_price': current_price,
|
||
|
|
'position_size': position,
|
||
|
|
'pnl': pnl,
|
||
|
|
'return_pct': pnl / (position * entry_price),
|
||
|
|
'exit_reason': exit_reason or "Signal exit"
|
||
|
|
})
|
||
|
|
|
||
|
|
position = 0
|
||
|
|
entry_price = 0
|
||
|
|
|
||
|
|
trades_df = pd.DataFrame(trades)
|
||
|
|
|
||
|
|
if trades_df.empty:
|
||
|
|
return {'trades': trades_df, 'total_return': 0, 'win_rate': 0, 'sharpe_ratio': 0}
|
||
|
|
|
||
|
|
# Calculate performance metrics
|
||
|
|
total_return = (capital - initial_capital) / initial_capital
|
||
|
|
win_rate = (trades_df['pnl'] > 0).mean()
|
||
|
|
returns = trades_df['return_pct']
|
||
|
|
sharpe_ratio = returns.mean() / returns.std() if returns.std() > 0 else 0
|
||
|
|
|
||
|
|
return {
|
||
|
|
'trades': trades_df,
|
||
|
|
'total_return': total_return,
|
||
|
|
'win_rate': win_rate,
|
||
|
|
'sharpe_ratio': sharpe_ratio,
|
||
|
|
'final_capital': capital,
|
||
|
|
'num_trades': len(trades_df),
|
||
|
|
'avg_return_per_trade': returns.mean(),
|
||
|
|
'max_loss': trades_df['pnl'].min(),
|
||
|
|
'max_gain': trades_df['pnl'].max()
|
||
|
|
}
|
||
|
|
|
||
|
|
class DrawdownVisualizer:
|
||
|
|
"""Visualization tools for conditional drawdown analysis"""
|
||
|
|
|
||
|
|
def __init__(self, analyzer: ConditionalDrawdownAnalyzer):
|
||
|
|
self.analyzer = analyzer
|
||
|
|
plt.style.use('seaborn-v0_8')
|
||
|
|
|
||
|
|
def plot_price_with_drawdowns(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_price: float,
|
||
|
|
recursive: bool = True,
|
||
|
|
figsize: Tuple[int, int] = (12, 8)):
|
||
|
|
"""Plot price series with drawdown analysis overlay"""
|
||
|
|
|
||
|
|
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=figsize, height_ratios=[3, 1])
|
||
|
|
|
||
|
|
# Main price plot
|
||
|
|
ax1.plot(price.index, price.values, 'b-', linewidth=1, alpha=0.7, label='Price')
|
||
|
|
ax1.axhline(y=entry_price, color='red', linestyle='--', alpha=0.8, label=f'Entry: ${entry_price:.2f}')
|
||
|
|
|
||
|
|
# Get drawdown metrics
|
||
|
|
if recursive:
|
||
|
|
metrics = self.analyzer.recursive_conditional_drawdown(price, entry_price)
|
||
|
|
else:
|
||
|
|
metrics = self.analyzer.basic_conditional_drawdown(price, entry_price)
|
||
|
|
|
||
|
|
# Draw confidence bands
|
||
|
|
ax1.axhline(y=entry_price - metrics.confidence_interval[0], color='orange',
|
||
|
|
linestyle=':', alpha=0.6, label=f'Expected DD: ${metrics.avg_drawdown:.2f}')
|
||
|
|
ax1.axhline(y=entry_price - metrics.confidence_interval[1], color='red',
|
||
|
|
linestyle=':', alpha=0.6, label=f'Max Expected: ${metrics.confidence_interval[1]:.2f}')
|
||
|
|
|
||
|
|
ax1.fill_between(price.index,
|
||
|
|
entry_price - metrics.confidence_interval[0],
|
||
|
|
entry_price - metrics.confidence_interval[1],
|
||
|
|
alpha=0.2, color='red', label='Risk Zone')
|
||
|
|
|
||
|
|
ax1.set_title(f'Price Analysis with Conditional Drawdown\n'
|
||
|
|
f'Avg DD: ${metrics.avg_drawdown:.2f} | Max DD: ${metrics.max_drawdown:.2f} | '
|
||
|
|
f'Recovery Rate: {metrics.recovery_probability:.1%}')
|
||
|
|
ax1.set_ylabel('Price ($)')
|
||
|
|
ax1.legend()
|
||
|
|
ax1.grid(True, alpha=0.3)
|
||
|
|
|
||
|
|
# Drawdown visualization
|
||
|
|
drawdown_series = price.apply(lambda x: max(0, entry_price - x))
|
||
|
|
ax2.fill_between(price.index, 0, drawdown_series,
|
||
|
|
where=(drawdown_series > 0), alpha=0.7, color='red', label='Current Drawdown')
|
||
|
|
ax2.axhline(y=metrics.avg_drawdown, color='orange', linestyle='-',
|
||
|
|
label=f'Average DD: ${metrics.avg_drawdown:.2f}')
|
||
|
|
ax2.set_ylabel('Drawdown ($)')
|
||
|
|
ax2.set_xlabel('Time')
|
||
|
|
ax2.legend()
|
||
|
|
ax2.grid(True, alpha=0.3)
|
||
|
|
|
||
|
|
plt.tight_layout()
|
||
|
|
return fig
|
||
|
|
|
||
|
|
def plot_drawdown_surface(self,
|
||
|
|
price: pd.Series,
|
||
|
|
surface: Optional[pd.DataFrame] = None,
|
||
|
|
figsize: Tuple[int, int] = (14, 10)):
|
||
|
|
"""Create comprehensive drawdown surface visualization"""
|
||
|
|
|
||
|
|
if surface is None:
|
||
|
|
surface = self.analyzer.create_drawdown_surface(price)
|
||
|
|
|
||
|
|
fig = plt.figure(figsize=figsize)
|
||
|
|
gs = fig.add_gridspec(2, 2, height_ratios=[1, 1], width_ratios=[3, 1])
|
||
|
|
|
||
|
|
# Main heatmap
|
||
|
|
ax1 = fig.add_subplot(gs[:, 0])
|
||
|
|
|
||
|
|
# Pivot for heatmap (if we have enough data points)
|
||
|
|
if len(surface) > 10:
|
||
|
|
# Create a simplified heatmap
|
||
|
|
pivot_data = surface.set_index('entry_price')[['avg_drawdown', 'recovery_probability']].T
|
||
|
|
|
||
|
|
im = ax1.imshow(pivot_data.values, aspect='auto', cmap='RdYlBu_r',
|
||
|
|
extent=[surface['entry_price'].min(), surface['entry_price'].max(), 0, 2])
|
||
|
|
|
||
|
|
ax1.set_xlabel('Entry Price ($)')
|
||
|
|
ax1.set_yticks([0.5, 1.5])
|
||
|
|
ax1.set_yticklabels(['Avg Drawdown', 'Recovery Prob'])
|
||
|
|
ax1.set_title('Drawdown Risk Surface')
|
||
|
|
|
||
|
|
# Add colorbar
|
||
|
|
cbar = plt.colorbar(im, ax=ax1)
|
||
|
|
cbar.set_label('Risk Level')
|
||
|
|
|
||
|
|
# Risk-reward scatter
|
||
|
|
ax2 = fig.add_subplot(gs[0, 1])
|
||
|
|
scatter = ax2.scatter(surface['avg_drawdown'], surface['recovery_probability'],
|
||
|
|
c=surface['entry_price'], s=30, alpha=0.7, cmap='viridis')
|
||
|
|
ax2.set_xlabel('Avg Drawdown')
|
||
|
|
ax2.set_ylabel('Recovery Probability')
|
||
|
|
ax2.set_title('Risk vs Recovery')
|
||
|
|
ax2.grid(True, alpha=0.3)
|
||
|
|
|
||
|
|
# Entry price distribution
|
||
|
|
ax3 = fig.add_subplot(gs[1, 1])
|
||
|
|
ax3.hist(surface['avg_drawdown'], bins=20, alpha=0.7, color='skyblue', edgecolor='black')
|
||
|
|
ax3.set_xlabel('Average Drawdown')
|
||
|
|
ax3.set_ylabel('Frequency')
|
||
|
|
ax3.set_title('Drawdown Distribution')
|
||
|
|
ax3.grid(True, alpha=0.3)
|
||
|
|
|
||
|
|
plt.tight_layout()
|
||
|
|
return fig
|
||
|
|
|
||
|
|
def plot_regime_analysis(self,
|
||
|
|
price: pd.Series,
|
||
|
|
entry_price: float,
|
||
|
|
current_time: Optional[pd.Timestamp] = None,
|
||
|
|
figsize: Tuple[int, int] = (12, 6)):
|
||
|
|
"""Plot regime detection analysis"""
|
||
|
|
|
||
|
|
if current_time is None:
|
||
|
|
current_time = price.index[-1]
|
||
|
|
|
||
|
|
# Calculate regime metrics over time
|
||
|
|
regime_data = []
|
||
|
|
lookback_window = min(50, len(price) // 4)
|
||
|
|
|
||
|
|
for i in range(lookback_window, len(price)):
|
||
|
|
subset = price.iloc[:i]
|
||
|
|
current_price = price.iloc[i]
|
||
|
|
current_dd = max(0, entry_price - current_price)
|
||
|
|
|
||
|
|
regime_info = self.analyzer.detect_abnormal_regime(subset, entry_price, current_dd)
|
||
|
|
regime_data.append({
|
||
|
|
'timestamp': price.index[i],
|
||
|
|
'severity_score': regime_info['severity_score'],
|
||
|
|
'is_abnormal': regime_info['is_abnormal_regime'],
|
||
|
|
'current_drawdown': current_dd,
|
||
|
|
'expected_drawdown': regime_info['expected_drawdown']
|
||
|
|
})
|
||
|
|
|
||
|
|
regime_df = pd.DataFrame(regime_data)
|
||
|
|
|
||
|
|
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=figsize, sharex=True)
|
||
|
|
|
||
|
|
# Price and regime zones
|
||
|
|
ax1.plot(price.index, price.values, 'b-', linewidth=1, label='Price')
|
||
|
|
ax1.axhline(y=entry_price, color='red', linestyle='--', label='Entry Price')
|
||
|
|
|
||
|
|
# Highlight abnormal periods
|
||
|
|
abnormal_periods = regime_df[regime_df['is_abnormal']]
|
||
|
|
if not abnormal_periods.empty:
|
||
|
|
for _, period in abnormal_periods.iterrows():
|
||
|
|
ax1.axvline(x=period['timestamp'], color='red', alpha=0.3)
|
||
|
|
|
||
|
|
ax1.set_ylabel('Price ($)')
|
||
|
|
ax1.set_title('Price Movement with Regime Detection')
|
||
|
|
ax1.legend()
|
||
|
|
ax1.grid(True, alpha=0.3)
|
||
|
|
|
||
|
|
# Severity score over time
|
||
|
|
ax2.plot(regime_df['timestamp'], regime_df['severity_score'],
|
||
|
|
'orange', linewidth=2, label='Severity Score')
|
||
|
|
ax2.axhline(y=1.0, color='green', linestyle='--', alpha=0.7, label='Normal')
|
||
|
|
ax2.axhline(y=2.0, color='red', linestyle='--', alpha=0.7, label='Abnormal Threshold')
|
||
|
|
|
||
|
|
# Shade abnormal regions
|
||
|
|
abnormal_mask = regime_df['severity_score'] > 2.0
|
||
|
|
ax2.fill_between(regime_df['timestamp'], 0, regime_df['severity_score'],
|
||
|
|
where=abnormal_mask, alpha=0.3, color='red', label='Abnormal Regime')
|
||
|
|
|
||
|
|
ax2.set_ylabel('Severity Score')
|
||
|
|
ax2.set_xlabel('Time')
|
||
|
|
ax2.set_title('Regime Abnormality Detection')
|
||
|
|
ax2.legend()
|
||
|
|
ax2.grid(True, alpha=0.3)
|
||
|
|
|
||
|
|
plt.tight_layout()
|
||
|
|
return fig
|
||
|
|
|
||
|
|
# Example usage and demonstration
|
||
|
|
def generate_sample_data(n_points: int = 1000, trend: float = 0.0005, volatility: float = 0.02) -> pd.Series:
|
||
|
|
"""Generate sample price data for testing"""
|
||
|
|
dates = pd.date_range(start='2023-01-01', periods=n_points, freq='H')
|
||
|
|
|
||
|
|
# Generate price path with trend and volatility
|
||
|
|
returns = np.random.normal(trend, volatility, n_points)
|
||
|
|
returns[100:150] = np.random.normal(-0.003, 0.04, 50) # Add a volatile period
|
||
|
|
returns[500:550] = np.random.normal(0.002, 0.01, 50) # Add a calm period
|
||
|
|
|
||
|
|
prices = 100 * np.exp(np.cumsum(returns))
|
||
|
|
return pd.Series(prices, index=dates)
|
||
|
|
|
||
|
|
def comprehensive_demo():
|
||
|
|
"""Run a comprehensive demonstration of the framework"""
|
||
|
|
print("🚀 Conditional Drawdown Framework - Comprehensive Demo\n")
|
||
|
|
|
||
|
|
# Generate sample data
|
||
|
|
print("📊 Generating sample price data...")
|
||
|
|
price_data = generate_sample_data(n_points=1000)
|
||
|
|
entry_price = price_data.iloc[100] # Enter after some history
|
||
|
|
|
||
|
|
# Initialize analyzer
|
||
|
|
analyzer = ConditionalDrawdownAnalyzer(confidence_level=0.95)
|
||
|
|
visualizer = DrawdownVisualizer(analyzer)
|
||
|
|
|
||
|
|
print(f"Entry Price: ${entry_price:.2f}")
|
||
|
|
print(f"Price Range: ${price_data.min():.2f} - ${price_data.max():.2f}\n")
|
||
|
|
|
||
|
|
# 1. Basic Analysis
|
||
|
|
print("🔍 1. Basic Conditional Drawdown Analysis")
|
||
|
|
basic_metrics = analyzer.basic_conditional_drawdown(price_data, entry_price)
|
||
|
|
print(f" Average Drawdown: ${basic_metrics.avg_drawdown:.2f}")
|
||
|
|
print(f" Maximum Drawdown: ${basic_metrics.max_drawdown:.2f}")
|
||
|
|
print(f" Recovery Probability: {basic_metrics.recovery_probability:.1%}")
|
||
|
|
print(f" Average Recovery Time: {basic_metrics.avg_recovery_time:.1f} periods\n")
|
||
|
|
|
||
|
|
# 2. Recursive Analysis
|
||
|
|
print("🔄 2. Recursive Conditional Drawdown Analysis")
|
||
|
|
recursive_metrics = analyzer.recursive_conditional_drawdown(price_data, entry_price)
|
||
|
|
print(f" Average Drawdown: ${recursive_metrics.avg_drawdown:.2f}")
|
||
|
|
print(f" Maximum Drawdown: ${recursive_metrics.max_drawdown:.2f}")
|
||
|
|
print(f" Recovery Probability: {recursive_metrics.recovery_probability:.1%}\n")
|
||
|
|
|
||
|
|
# 3. Volatility Adjustment
|
||
|
|
print("📈 3. Volatility-Adjusted Analysis")
|
||
|
|
vol_result = analyzer.volatility_adjusted_drawdown(price_data, entry_price)
|
||
|
|
print(f" Volatility Coefficient: {vol_result['volatility_coefficient']:.3f}")
|
||
|
|
print(f" Adjusted Avg Drawdown: ${vol_result['adjusted_metrics'].avg_drawdown:.2f}")
|
||
|
|
print(f" Base vs Adjusted: ${vol_result['base_metrics'].avg_drawdown:.2f} → "
|
||
|
|
f"${vol_result['adjusted_metrics'].avg_drawdown:.2f}\n")
|
||
|
|
|
||
|
|
# 4. Order Book Integration
|
||
|
|
print("📋 4. Order Book Integration")
|
||
|
|
ob_result = analyzer.order_book_adjusted_drawdown(price_data, entry_price)
|
||
|
|
print(f" Liquidity Adjustment Factor: {ob_result['liquidity_adjustment_factor']:.3f}")
|
||
|
|
print(f" Final Adjusted Drawdown: ${ob_result['liquidity_adjusted_metrics'].avg_drawdown:.2f}")
|
||
|
|
print(f" Order Book - Bid Depth: {ob_result['order_book'].bid_depth:.0f}, "
|
||
|
|
f"Spread: ${ob_result['order_book'].spread:.4f}\n")
|
||
|
|
|
||
|
|
# 5. Drawdown Surface
|
||
|
|
print("🗺️ 5. Creating Drawdown Surface...")
|
||
|
|
surface = analyzer.create_drawdown_surface(price_data, resolution=0.01)
|
||
|
|
print(f" Surface calculated for {len(surface)} entry points")
|
||
|
|
print(f" Min Risk Entry: ${surface.loc[surface['avg_drawdown'].idxmin(), 'entry_price']:.2f} "
|
||
|
|
f"(DD: ${surface['avg_drawdown'].min():.2f})")
|
||
|
|
print(f" Max Risk Entry: ${surface.loc[surface['avg_drawdown'].idxmax(), 'entry_price']:.2f} "
|
||
|
|
f"(DD: ${surface['avg_drawdown'].max():.2f})\n")
|
||
|
|
|
||
|
|
# 6. Entry Optimization
|
||
|
|
print("🎯 6. Entry Optimization")
|
||
|
|
optimal_entries = analyzer.optimize_entries(price_data, reward_target=0.03, max_entries=5)
|
||
|
|
if not optimal_entries.empty:
|
||
|
|
print(" Top 5 Optimal Entry Points:")
|
||
|
|
for i, row in optimal_entries.iterrows():
|
||
|
|
print(f" ${row['entry_price']:.2f} - Risk/Reward: {row['risk_reward_ratio']:.2f}, "
|
||
|
|
f"Score: {row['score']:.3f}")
|
||
|
|
print()
|
||
|
|
|
||
|
|
# 7. Regime Detection
|
||
|
|
print("🚨 7. Regime Detection")
|
||
|
|
current_price = price_data.iloc[-1]
|
||
|
|
current_drawdown = max(0, entry_price - current_price)
|
||
|
|
regime_info = analyzer.detect_abnormal_regime(price_data, entry_price, current_drawdown)
|
||
|
|
print(f" Current Drawdown: ${current_drawdown:.2f}")
|
||
|
|
print(f" Expected Drawdown: ${regime_info['expected_drawdown']:.2f}")
|
||
|
|
print(f" Severity Score: {regime_info['severity_score']:.2f}")
|
||
|
|
print(f" Abnormal Regime: {regime_info['is_abnormal_regime']}")
|
||
|
|
print(f" Recommendation: {regime_info['recommendation']}\n")
|
||
|
|
|
||
|
|
# 8. Backtesting
|
||
|
|
print("📊 8. Strategy Backtesting")
|
||
|
|
# Create simple entry signals (random for demo)
|
||
|
|
np.random.seed(42)
|
||
|
|
signals = pd.Series(np.random.choice([0, 1, -1], size=len(price_data), p=[0.8, 0.1, 0.1]),
|
||
|
|
index=price_data.index)
|
||
|
|
|
||
|
|
backtest_result = analyzer.backtest_strategy(price_data, signals, exit_method='adaptive_stop')
|
||
|
|
print(f" Total Return: {backtest_result['total_return']:.1%}")
|
||
|
|
print(f" Win Rate: {backtest_result['win_rate']:.1%}")
|
||
|
|
print(f" Sharpe Ratio: {backtest_result['sharpe_ratio']:.2f}")
|
||
|
|
print(f" Number of Trades: {backtest_result['num_trades']}")
|
||
|
|
print(f" Average Return per Trade: {backtest_result['avg_return_per_trade']:.1%}\n")
|
||
|
|
|
||
|
|
print("✅ Framework demonstration complete!")
|
||
|
|
print("📈 Use the visualizer to create plots:")
|
||
|
|
print(" - visualizer.plot_price_with_drawdowns(price_data, entry_price)")
|
||
|
|
print(" - visualizer.plot_drawdown_surface(price_data)")
|
||
|
|
print(" - visualizer.plot_regime_analysis(price_data, entry_price)")
|
||
|
|
|
||
|
|
return {
|
||
|
|
'price_data': price_data,
|
||
|
|
'entry_price': entry_price,
|
||
|
|
'analyzer': analyzer,
|
||
|
|
'visualizer': visualizer,
|
||
|
|
'surface': surface,
|
||
|
|
'backtest_result': backtest_result
|
||
|
|
}
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
# Run the comprehensive demo
|
||
|
|
demo_results = comprehensive_demo()
|