diff --git a/model.py b/model.py index 96bca4d..eb9cfe8 100644 --- a/model.py +++ b/model.py @@ -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