Use market fundamentals intsead of empirical era adjustments.
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user