Add adaptive volatility window.
This commit is contained in:
@@ -152,36 +152,146 @@ def analyze_trends(df):
|
||||
return smoothed_returns
|
||||
|
||||
|
||||
def calculate_volatility(df, short_window=30, medium_window=90, long_window=180):
|
||||
def calculate_adaptive_volatility(
|
||||
df,
|
||||
short_window=30,
|
||||
medium_window=90,
|
||||
long_window=180,
|
||||
vol_clip_min=0.5,
|
||||
vol_clip_max=2.0,
|
||||
):
|
||||
"""
|
||||
Final volatility calculation with precise period adjustments.
|
||||
Calculate volatility with adaptive window sizes based on market conditions.
|
||||
Returns a single volatility value for the most recent period.
|
||||
"""
|
||||
df = df.copy()
|
||||
if "Log_Return" not in df.columns:
|
||||
df["Log_Price"] = np.log(df["Close"])
|
||||
df["Log_Return"] = df["Log_Price"].diff()
|
||||
df["Log_Return"] = np.log(df["Close"]).diff()
|
||||
|
||||
# Base volatility calculation
|
||||
short_vol = df["Log_Return"].ewm(span=short_window, adjust=False).std().iloc[-1]
|
||||
medium_vol = df["Log_Return"].ewm(span=medium_window, adjust=False).std().iloc[-1]
|
||||
long_vol = df["Log_Return"].ewm(span=long_window, adjust=False).std().iloc[-1]
|
||||
# Remove any NaN values that could cause issues
|
||||
df = df.dropna()
|
||||
|
||||
# Standard weights
|
||||
base_vol = 0.2 * short_vol + 0.5 * medium_vol + 0.3 * long_vol
|
||||
if len(df) < long_window:
|
||||
# Not enough data, fall back to simple volatility
|
||||
return df["Log_Return"].std()
|
||||
|
||||
# Precise period-specific scaling
|
||||
start_date = df["Date"].min()
|
||||
if start_date >= pd.Timestamp("2020-01-01"):
|
||||
base_adjustment = 0.64 # Slightly increased
|
||||
elif start_date >= pd.Timestamp("2016-07-09"):
|
||||
base_adjustment = 0.67 # Slightly increased
|
||||
elif start_date >= pd.Timestamp("2015-01-01"):
|
||||
base_adjustment = 0.69 # Increased for mid period
|
||||
else:
|
||||
# Early period with less aggressive scaling
|
||||
base_adjustment = 0.70 # Fixed value for stability
|
||||
# Get recent data for efficiency
|
||||
lookback = max(long_window * 2, 360) # Use enough data for stable estimates
|
||||
recent_df = df.iloc[-lookback:].copy() if len(df) > lookback else df.copy()
|
||||
|
||||
return base_vol * base_adjustment
|
||||
try:
|
||||
# Initial volatility estimate using base windows
|
||||
short_vol = recent_df["Log_Return"].ewm(span=short_window, adjust=False).std()
|
||||
medium_vol = recent_df["Log_Return"].ewm(span=medium_window, adjust=False).std()
|
||||
long_vol = recent_df["Log_Return"].ewm(span=long_window, adjust=False).std()
|
||||
|
||||
# Ensure we have valid volatility values
|
||||
if short_vol.iloc[-1] == 0 or np.isnan(short_vol.iloc[-1]):
|
||||
return df["Log_Return"].std() # Fallback to simple volatility
|
||||
|
||||
# Calculate regime indicators for recent period
|
||||
medium_vol_mean = medium_vol.rolling(min(90, len(recent_df))).mean()
|
||||
if medium_vol_mean.iloc[-1] == 0:
|
||||
vol_regime = pd.Series([1.0] * len(recent_df))
|
||||
else:
|
||||
vol_regime = short_vol / medium_vol_mean
|
||||
|
||||
vol_regime = vol_regime.clip(vol_clip_min, vol_clip_max)
|
||||
|
||||
# Get most recent regime reading
|
||||
latest_regime = vol_regime.iloc[-1]
|
||||
|
||||
# Adjust window sizes based on current regime
|
||||
adj_factor = 1 / latest_regime
|
||||
adj_short = max(10, int(short_window * adj_factor)) # Minimum window of 10
|
||||
adj_medium = max(30, int(medium_window * adj_factor))
|
||||
adj_long = max(60, int(long_window * adj_factor))
|
||||
|
||||
# Calculate final volatilities using adjusted windows
|
||||
final_short = recent_df["Log_Return"].iloc[-adj_short:].std()
|
||||
final_medium = recent_df["Log_Return"].iloc[-adj_medium:].std()
|
||||
final_long = recent_df["Log_Return"].iloc[-adj_long:].std()
|
||||
|
||||
# If any volatility measure is NaN or 0, fall back to simple volatility
|
||||
if np.isnan([final_short, final_medium, final_long]).any() or 0 in [
|
||||
final_short,
|
||||
final_medium,
|
||||
final_long,
|
||||
]:
|
||||
return df["Log_Return"].std()
|
||||
|
||||
# Calculate regime-based weights
|
||||
high_vol_weight = (latest_regime - vol_clip_min) / (vol_clip_max - vol_clip_min)
|
||||
base_weights = np.array([0.2, 0.5, 0.3])
|
||||
stress_weights = np.array([0.4, 0.4, 0.2])
|
||||
|
||||
# Interpolate between base and stress weights
|
||||
weights = (
|
||||
base_weights * (1 - high_vol_weight) + stress_weights * high_vol_weight
|
||||
)
|
||||
|
||||
# Calculate final volatility
|
||||
final_vol = (
|
||||
final_short * weights[0]
|
||||
+ final_medium * weights[1]
|
||||
+ final_long * weights[2]
|
||||
)
|
||||
|
||||
# Add uncertainty adjustment
|
||||
regime_change = abs(vol_regime.diff()).fillna(0)
|
||||
regime_change_mean = regime_change.rolling(5, min_periods=1).mean().iloc[-1]
|
||||
if regime_change_mean == 0:
|
||||
uncertainty_adjustment = 1.0
|
||||
else:
|
||||
regime_change_zscore = regime_change.iloc[-1] / regime_change_mean
|
||||
uncertainty_adjustment = 1 + np.clip(regime_change_zscore / 2, 0, 0.5)
|
||||
|
||||
return max(final_vol * uncertainty_adjustment, df["Log_Return"].std() * 0.5)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in adaptive volatility calculation: {e}")
|
||||
# Fall back to simple volatility calculation
|
||||
return df["Log_Return"].std()
|
||||
|
||||
|
||||
def calculate_volatility(df, short_window=30, medium_window=90, long_window=180):
|
||||
"""
|
||||
Calculate volatility using adaptive windows and era-specific adjustments.
|
||||
Returns a single volatility value.
|
||||
"""
|
||||
if len(df) < 30:
|
||||
return 0.02 # Return a reasonable default for very short periods
|
||||
|
||||
try:
|
||||
# Calculate adaptive volatility
|
||||
base_vol = calculate_adaptive_volatility(
|
||||
df,
|
||||
short_window=short_window,
|
||||
medium_window=medium_window,
|
||||
long_window=long_window,
|
||||
)
|
||||
|
||||
if np.isnan(base_vol) or base_vol == 0:
|
||||
base_vol = df["Close"].pct_change().std()
|
||||
|
||||
# Era-specific adjustments
|
||||
start_date = df["Date"].min()
|
||||
if start_date >= pd.Timestamp("2020-01-01"):
|
||||
base_adjustment = 0.64
|
||||
elif start_date >= pd.Timestamp("2016-07-09"):
|
||||
base_adjustment = 0.67
|
||||
elif start_date >= pd.Timestamp("2015-01-01"):
|
||||
base_adjustment = 0.69
|
||||
else:
|
||||
base_adjustment = 0.70
|
||||
|
||||
return max(
|
||||
base_vol * base_adjustment, 0.01
|
||||
) # Ensure we never return 0 volatility
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error in volatility calculation: {e}")
|
||||
# Fall back to simple volatility with minimum floor
|
||||
return max(df["Close"].pct_change().std(), 0.01)
|
||||
|
||||
|
||||
# Era definitions
|
||||
|
||||
Reference in New Issue
Block a user