From 96fdf5a88e55f38b136215f3048946d3db968a9c Mon Sep 17 00:00:00 2001 From: Sam Fredrickson Date: Tue, 19 Nov 2024 07:58:40 -0800 Subject: [PATCH] Use market fundamentals intsead of empirical era adjustments. --- model.py | 347 ++++++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 256 insertions(+), 91 deletions(-) diff --git a/model.py b/model.py index bee0242..2558eab 100644 --- a/model.py +++ b/model.py @@ -151,6 +151,168 @@ def get_nice_price_points(min_price, max_price): return np.array(price_points) +# Market metrics + + +class MarketFundamentals: + """ + Calculate and track fundamental market metrics for Bitcoin. + Designed to be extensible for additional metrics. + """ + + def __init__(self): + # Constants + self.GENESIS_DATE = pd.Timestamp("2009-01-03") + self.BLOCKS_PER_DAY = 144 + self.HALVING_INTERVAL = 210000 # blocks + + # Volatility adjustment parameters (from our tuning) + self.VOLUME_SCALE = 70 + self.DEPTH_SCALE = 7 + self.BASE_ADJUSTMENT = 0.68 + + def calculate_total_supply(self, date): + """Calculate total Bitcoin supply at a given date.""" + days_since_genesis = (date - self.GENESIS_DATE).days + if days_since_genesis < 0: + return 0 + + total_supply = 0 + remaining_blocks = days_since_genesis * self.BLOCKS_PER_DAY + current_reward = 50 + + while remaining_blocks > 0 and current_reward >= 0.01: + blocks_at_this_reward = min(remaining_blocks, self.HALVING_INTERVAL) + total_supply += blocks_at_this_reward * current_reward + remaining_blocks -= blocks_at_this_reward + current_reward /= 2 + + # Adjust for missed blocks and lost coins + total_supply *= 0.95 # Account for varying block times + total_supply *= 0.93 # Estimate for lost/inaccessible coins + + return total_supply + + def get_block_reward(self, date): + """Get Bitcoin block reward at a given date.""" + days_since_genesis = (date - self.GENESIS_DATE).days + if days_since_genesis < 0: + return 0 + + halvings = days_since_genesis // (4 * 365) # Approximate halving periods + return 50 / (2**halvings) + + def calculate_supply_metrics(self, date): + """Calculate supply-related metrics.""" + total_supply = self.calculate_total_supply(date) + block_reward = self.get_block_reward(date) + daily_new_supply = block_reward * self.BLOCKS_PER_DAY + + return { + "total_supply": total_supply, + "daily_new_supply": daily_new_supply, + "supply_growth_rate": daily_new_supply / total_supply, + "stock_to_flow": total_supply / (daily_new_supply * 365), # Annualized + } + + def calculate_market_metrics(self, df, date, window=30): + """Calculate market activity metrics.""" + recent_data = df[df["Date"] <= date].tail(window) + + if len(recent_data) < window: + return {"avg_volume": 0, "price_volatility": 0, "price_impact": 0} + + avg_volume = recent_data["Volume"].mean() + price_volatility = recent_data["Close"].pct_change().std() + price_impact = recent_data["Close"].std() / recent_data["Close"].mean() + + return { + "avg_volume": avg_volume, + "price_volatility": price_volatility, + "price_impact": price_impact, + } + + def get_market_maturity_metrics(self, df, date, window=30): + """ + Combine supply and market metrics to assess market maturity. + """ + supply_metrics = self.calculate_supply_metrics(date) + market_metrics = self.calculate_market_metrics(df, date, window) + + # Calculate combined metrics + volume_to_supply = market_metrics["avg_volume"] / supply_metrics["total_supply"] + market_depth = volume_to_supply / (market_metrics["price_impact"] + 0.001) + + return { + "volume_to_supply": volume_to_supply, + "supply_growth_rate": supply_metrics["supply_growth_rate"], + "market_depth": market_depth, + "stock_to_flow": supply_metrics["stock_to_flow"], + "price_impact": market_metrics["price_impact"], + } + + def calculate_volatility_adjustment(self, metrics): + """ + Calculate volatility adjustment based on market metrics. + """ + # Supply-based component + supply_based_vol = np.sqrt(metrics["supply_growth_rate"] * 365 * 100) + + # Market maturity component + maturity_factor = 1 - np.clip( + metrics["volume_to_supply"] * self.VOLUME_SCALE, 0, 0.6 + ) + + # Market depth component + depth_factor = np.clip( + 1 / np.sqrt(1 + metrics["market_depth"] * self.DEPTH_SCALE), 0.7, 1.3 + ) + + # Combine factors + adjustment = ( + self.BASE_ADJUSTMENT + * (1 + supply_based_vol) + * maturity_factor + * depth_factor + ) + + # Ensure reasonable bounds + return np.clip(adjustment, 0.65, 0.75) + + +def compare_adjustments(df, fundamentals): + """ + Compare fundamental-based adjustments with original era-based ones. + """ + # Sample dates for comparison + date_range = pd.date_range(start=df["Date"].min(), end=df["Date"].max(), freq="30D") + + results = [] + for date in date_range: + # Calculate era-based adjustment + if date < pd.Timestamp("2017-12-10"): + era_adj = 0.71 # early era + elif date < pd.Timestamp("2020-01-01"): + era_adj = 0.69 # transition era + else: + era_adj = 0.67 # mature era + + # Calculate fundamental-based adjustment + metrics = fundamentals.get_market_maturity_metrics(df, date) + fund_adj = fundamentals.calculate_volatility_adjustment(metrics) + + results.append( + { + "date": date, + "era_adjustment": era_adj, + "fundamental_adjustment": fund_adj, + "metrics": metrics, + } + ) + + return pd.DataFrame(results) + + # Analysis functions @@ -318,70 +480,83 @@ def calculate_adaptive_volatility( 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. - """ + """Calculate volatility using fundamental metrics and adaptive windows.""" + df = df.copy() + df["Log_Return"] = np.log(df["Close"]).diff() + if len(df) < 30: - return 0.02 # Return a reasonable default for very short periods + return 0.02 # 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, + # Initialize fundamentals calculator + fundamentals = MarketFundamentals() + current_date = df["Date"].max() + + # Get recent data for efficiency + lookback = max(long_window * 2, 360) + recent_df = df.iloc[-lookback:].copy() if len(df) > lookback else df.copy() + + # Calculate base volatilities + 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() + + if short_vol.iloc[-1] == 0 or np.isnan(short_vol.iloc[-1]): + return df["Log_Return"].std() + + # Calculate volatility regime indicators + medium_vol_mean = medium_vol.rolling(min(90, len(recent_df))).mean() + long_vol_mean = long_vol.rolling(min(180, len(recent_df))).mean() + + # Compare short-term to both medium and long-term volatility + if medium_vol_mean.iloc[-1] == 0: + vol_regime = pd.Series([1.0] * len(recent_df)) + else: + medium_regime = short_vol / medium_vol_mean + long_regime = short_vol / long_vol_mean + vol_regime = pd.concat([medium_regime, long_regime], axis=1).max(axis=1) + + vol_regime = vol_regime.clip(0.5, 2.0) + latest_regime = vol_regime.iloc[-1] + + # Get market metrics + metrics = fundamentals.get_market_maturity_metrics(df, current_date) + + # Calculate adaptive weights + high_vol_weight = (latest_regime - 0.5) / 1.5 # 1.5 = 2.0 - 0.5 + base_weights = np.array([0.2, 0.5, 0.3]) + stress_weights = np.array([0.4, 0.4, 0.2]) + weights = ( + base_weights * (1 - high_vol_weight) + stress_weights * high_vol_weight ) - if np.isnan(base_vol) or base_vol == 0: - base_vol = df["Close"].pct_change().std() + # Calculate final volatilities + final_short = recent_df["Log_Return"].iloc[-short_window:].std() + final_medium = recent_df["Log_Return"].iloc[-medium_window:].std() + final_long = recent_df["Log_Return"].iloc[-long_window:].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 + if np.isnan([final_short, final_medium, final_long]).any() or 0 in [ + final_short, + final_medium, + final_long, + ]: + return df["Log_Return"].std() - return max( - base_vol * base_adjustment, 0.01 - ) # Ensure we never return 0 volatility + # Apply market-based adjustment + market_adjustment = fundamentals.calculate_volatility_adjustment(metrics) + + # Calculate final volatility + final_vol = ( + final_short * weights[0] + + final_medium * weights[1] + + final_long * weights[2] + ) * market_adjustment + + return max(final_vol, df["Log_Return"].std() * 0.5) 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 -era_adjustments = { - "early": { - "start_date": pd.Timestamp("2013-01-01"), - "end_date": pd.Timestamp("2017-12-10"), - "volatility_scale": 0.71, # Slight increase - "trend_scale": 0.75, - "skew_scale": 1.0, - }, - "transition": { - "start_date": pd.Timestamp("2017-12-10"), - "end_date": pd.Timestamp("2020-01-01"), - "volatility_scale": 0.69, # Slight increase - "trend_scale": 0.80, - "skew_scale": 1.0, - }, - "mature": { - "start_date": pd.Timestamp("2020-01-01"), - "end_date": pd.Timestamp("2100-01-01"), - "volatility_scale": 0.67, # Slight increase - "trend_scale": 0.85, - "skew_scale": 1.0, - }, -} + return df["Log_Return"].std() def adjust_trend_expectations(expected_returns, cycle_position): @@ -518,10 +693,10 @@ def calculate_confidence_intervals(simulated_paths, confidence_levels=[0.95, 0.6 def project_prices( df, days_forward=365, simulations=1000, confidence_levels=[0.95, 0.68] ): - """ - Modified projection function incorporating enhanced uncertainty estimation. - """ + """Generate price projections with fundamental-based adjustments.""" df = df.copy() + fundamentals = MarketFundamentals() + df["Log_Price"] = np.log(df["Close"]) df["Log_Return"] = df["Log_Price"].diff() @@ -535,43 +710,24 @@ def project_prices( last_price = df["Close"].iloc[-1] last_date = df["Date"].iloc[-1] - # Generate dates for projection + # Generate projection dates future_dates = pd.date_range( start=last_date + timedelta(days=1), periods=days_forward, freq="D" ) - # Calculate expected returns with cycle boundary handling + # Calculate expected returns future_cycle_days = [ (current_cycle_days + i) % (4 * 365) for i in range(days_forward) ] cycle_trends = analyze_trends(df) - - # Get base expected returns expected_returns = np.array( [cycle_trends.get(day, cycle_trends.mean()) for day in future_cycle_days] ) - # Apply trend adjustments - expected_returns = adjust_trend_expectations(expected_returns, cycle_position) - # Calculate base volatility base_volatility = calculate_volatility(df) - # Get era adjustments - current_era = None - for era, params in era_adjustments.items(): - if ( - df["Date"].min() >= params["start_date"] - and df["Date"].min() < params["end_date"] - ): - current_era = era - break - if current_era is None: - current_era = "mature" - - era_params = era_adjustments[current_era] - - # Get projection adjustments with market awareness + # Get projection adjustments projection_adjustments = get_projection_adjustments( days_forward, cycle_position, df ) @@ -581,30 +737,39 @@ def project_prices( simulated_paths = np.zeros((days_forward, simulations)) for sim in range(simulations): - # Apply era-specific adjustments - drift = expected_returns * era_params["trend_scale"] - vol = base_volatility * era_params["volatility_scale"] - - # Scale volatility by projection adjustments + drift = expected_returns + vol = base_volatility time_scaled_vol = vol * projection_adjustments - # Generate returns with time-varying volatility returns = np.random.normal(loc=drift, scale=time_scaled_vol, size=days_forward) - # Calculate price path cumulative_returns = np.cumsum(returns) price_path = last_price * np.exp(cumulative_returns) simulated_paths[:, sim] = price_path - # Calculate results with dynamic confidence intervals + # Calculate results results = pd.DataFrame(index=future_dates) results["Median"] = np.percentile(simulated_paths, 50, axis=1) results["Expected_Trend"] = last_price * np.exp(np.cumsum(drift)) - # Calculate confidence intervals with dynamic adjustment - ci_results = calculate_confidence_intervals(simulated_paths, confidence_levels) - for key, values in ci_results.items(): - results[key] = values + # Calculate confidence intervals + for level in confidence_levels: + # Get market metrics for confidence interval adjustment + metrics = fundamentals.get_market_maturity_metrics(df, current_date) + maturity_adjustment = np.clip(metrics["market_depth"], 0, 0.5) + + # Calculate adjusted confidence level + effective_level = level + (1 - level) * maturity_adjustment + + lower_percentile = (1 - effective_level) * 100 / 2 + upper_percentile = 100 - lower_percentile + + results[f"Lower_{int(level*100)}"] = np.percentile( + simulated_paths, lower_percentile, axis=1 + ) + results[f"Upper_{int(level*100)}"] = np.percentile( + simulated_paths, upper_percentile, axis=1 + ) return results