sam/bitcoin-model
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tuning session; removed market maturity.

Browse Source 
The market maturity score only complicated the model with no clear
benefit. Still working on getting the various backtests tuned.
This commit is contained in:
sam
2024-11-15 18:07:33 -08:00
parent b9cf04aed7
commit e484331196
2 changed files with 201 additions and 295 deletions
Download Patch File Download Diff File Expand all files Collapse all files
model.py
+138 -279
View File
@@ -106,7 +106,7 @@ def get_nice_price_points(min_price, max_price):
def analyze_trends(df):
"""
Analyze Bitcoin price trends using log returns with simple moving average smoothing.
Analyze Bitcoin price trends using log returns with asymmetric dampening.
"""
df = df.copy()
@@ -128,265 +128,140 @@ def analyze_trends(df):
window = 60
smoothed_returns = position_returns.rolling(
window=window,
center=True, # Center the window for better trend capture
min_periods=int(window / 2), # Allow partial windows to reduce edge effects
center=True,
min_periods=int(window / 2),
).mean()
# Fill any NaN values at the edges
smoothed_returns = smoothed_returns.fillna(method="bfill").fillna(method="ffill")
# Apply asymmetric dampening to extreme values
returns_std = smoothed_returns.std()
def asymmetric_dampen(x):
if x > 2 * returns_std:
return x * 0.6 # Stronger dampening for positive extremes
elif x < -2 * returns_std:
return x * 0.7 # Slightly less dampening for negative extremes
return x
smoothed_returns = smoothed_returns.map(asymmetric_dampen)
# Additional dampening based on absolute magnitude
magnitude_factor = 0.9 # Global dampening factor
smoothed_returns = smoothed_returns * magnitude_factor
return smoothed_returns
def calculate_volatility(df, short_window=30, medium_window=90, long_window=180):
"""
Calculate volatility using multiple timeframes and exponential weighting.
Returns a more nuanced estimate of current market volatility.
Final volatility calculation with precise period adjustments.
"""
df = df.copy()
# Calculate log returns if not already present
if "Log_Return" not in df.columns:
df["Log_Price"] = np.log(df["Close"])
df["Log_Return"] = df["Log_Price"].diff()
# Calculate exponentially weighted volatilities for different timeframes
short_vol = df["Log_Return"].ewm(span=short_window).std().iloc[-1]
medium_vol = df["Log_Return"].ewm(span=medium_window).std().iloc[-1]
long_vol = df["Log_Return"].ewm(span=long_window).std().iloc[-1]
# 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]
# Blend the estimates with more weight on recent data
base_vol = 0.5 * short_vol + 0.3 * medium_vol + 0.2 * long_vol
# Standard weights
base_vol = 0.2 * short_vol + 0.5 * medium_vol + 0.3 * long_vol
# Scale up volatility to target ~68% coverage
volatility_scale = 1.2
return base_vol * volatility_scale
# 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
return base_vol * base_adjustment
def calculate_market_maturity_score(df):
# 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,
},
}
def adjust_trend_expectations(expected_returns, cycle_position):
"""
Calculate a market maturity score (0-1) based on multiple indicators.
Higher scores indicate a more mature market.
Simple trend adjustment.
"""
df = df.copy()
if cycle_position > 0.75:
damping_factor = 0.70
else:
damping_factor = 0.85
# 1. Enhanced volume-based metrics
# Use rolling median instead of mean to reduce impact of outliers
df["volume_ma90"] = df["Volume"].rolling(window=90).median()
df["volume_ma365"] = df["Volume"].rolling(window=365).median()
# Calculate relative volume growth using log differences
# This better handles exponential growth in volume over time
volume_growth_90d = np.log(df["volume_ma90"] / df["volume_ma90"].shift(90)).fillna(
0
)
volume_growth_365d = np.log(
df["volume_ma365"] / df["volume_ma365"].shift(365)
).fillna(0)
# Normalize volume growth to rolling volatility of volume
# This adapts to different market epochs
volume_growth_std_90 = volume_growth_90d.rolling(window=90).std()
volume_growth_std_365 = volume_growth_365d.rolling(window=365).std()
normalized_volume_growth = (
(volume_growth_90d / volume_growth_std_90).clip(-2, 2) * 0.4
+ (volume_growth_365d / volume_growth_std_365).clip(-2, 2) * 0.6
).fillna(0)
# Transform to 0-1 scale using sigmoid function
volume_score = 1 / (1 + np.exp(-normalized_volume_growth))
# 2. Volatility maturity (lower volatility = more mature)
df["rolling_vol_90"] = df["Daily_Return"].rolling(window=90).std() * np.sqrt(365)
df["rolling_vol_365"] = df["Daily_Return"].rolling(window=365).std() * np.sqrt(365)
# Normalize volatility relative to its historical range
vol_score_90 = 1 / (
1 + df["rolling_vol_90"] / df["rolling_vol_90"].rolling(window=365).median()
)
vol_score_365 = 1 / (
1 + df["rolling_vol_365"] / df["rolling_vol_365"].rolling(window=730).median()
)
vol_maturity = vol_score_90 * 0.4 + vol_score_365 * 0.6
# 3. Market efficiency score using multiple timeframes
efficiency_scores = []
for window in [30, 90]:
# Calculate absolute autocorrelation at multiple lags
for lag in [1, 2, 3, 5]:
autocorr = (
df["Daily_Return"]
.rolling(window=window)
.apply(lambda x: abs(pd.Series(x).autocorr(lag)))
)
efficiency_scores.append(1 - autocorr)
efficiency = pd.concat(efficiency_scores, axis=1).mean(axis=1)
# 4. Futures market impact (post-2017)
futures_date = pd.Timestamp("2017-12-10")
futures_impact = (df["Date"] > futures_date).astype(float)
# Progressive futures market maturation
days_since_futures = (df["Date"] - futures_date).dt.total_seconds() / (24 * 60 * 60)
futures_maturity = futures_impact * (1 - np.exp(-days_since_futures / 365))
# Combine scores with dynamic weights
base_weights = {
"volume": 0.25,
"volatility": 0.30,
"efficiency": 0.25,
"futures": 0.20,
}
# Adjust weights based on data availability
lookback = pd.Timestamp("2016-01-01")
historical_period = (df["Date"] < lookback).astype(float)
# Add stronger early-market adjustment
if df["Date"].min() < pd.Timestamp("2013-01-01"):
historical_period *= 1.5 # Increase uncertainty for pre-2013 data
# Reduce weight of futures impact for historical data
weights = base_weights.copy()
weights["futures"] = weights["futures"] * (1 - historical_period)
# Redistribute futures weight to other components in historical period
historical_adjustment = (weights["futures"] * historical_period) / 3
weights["volume"] += historical_adjustment
weights["volatility"] += historical_adjustment
weights["efficiency"] += historical_adjustment
# Calculate final score
maturity_score = (
weights["volume"] * volume_score
+ weights["volatility"] * vol_maturity
+ weights["efficiency"] * efficiency
+ weights["futures"] * futures_maturity
)
# Apply non-linear transformation to better distinguish maturity levels
maturity_score = 1 / (1 + np.exp(-4 * (maturity_score - 0.5)))
# Final smoothing
maturity_score = maturity_score.rolling(
window=30, min_periods=1, center=True
).mean()
return maturity_score
return expected_returns * damping_factor
def adjust_projections_for_maturity(df, projections, maturity_score):
def get_projection_adjustments(days_forward, current_cycle_position):
"""
Adjust price projections based on market maturity score.
More mature markets should have tighter confidence intervals
and more conservative growth expectations.
Final projection adjustments with precise uncertainty scaling.
"""
# Get final maturity score
final_maturity = maturity_score.iloc[-1]
adjustments = np.ones(days_forward)
# Adjust confidence intervals based on maturity
# More mature markets = tighter intervals
ci_adjustment = 1 - (final_maturity * 0.3) # Max 30% reduction in interval width
# Fixed base uncertainty with slight cycle variation
base_uncertainty = 0.016 # Standard rate
if current_cycle_position > 0.75:
base_uncertainty *= 1.1 # 10% increase late cycle
# Reduce conservatism in high-maturity periods
if final_maturity > 0.7:
ci_adjustment *= 1.2 # Widen intervals less in very mature periods
for i in range(days_forward):
# Conservative growth with fixed cap
time_factor = min(1 + (i / 365) * base_uncertainty, 1.055) # Lower cap
# Adjust expected returns based on maturity
# More mature markets = more conservative growth
returns_adjustment = 1 - (
final_maturity * 0.2
) # Max 20% reduction in expected returns
# Simpler cycle factors
cycle_position = (current_cycle_position + i / 1460) % 1
if cycle_position > 0.75:
cycle_factor = 0.94
else:
cycle_factor = 0.96
adjusted_projections = projections.copy()
adjustments[i] = time_factor * cycle_factor
# Adjust confidence intervals
for ci in [68, 95]:
upper_key = f"Upper_{ci}"
lower_key = f"Lower_{ci}"
median = adjusted_projections["Median"]
# Calculate distances from median
upper_distance = adjusted_projections[upper_key] - median
lower_distance = median - adjusted_projections[lower_key]
# Apply maturity-based adjustment
adjusted_projections[upper_key] = median + (upper_distance * ci_adjustment)
adjusted_projections[lower_key] = median - (lower_distance * ci_adjustment)
# Adjust expected trend
trend_distance = (
adjusted_projections["Expected_Trend"] - adjusted_projections["Median"]
)
adjusted_projections["Expected_Trend"] = (
adjusted_projections["Median"] + trend_distance * returns_adjustment
)
return adjusted_projections
return adjustments
def project_prices(
df, days_forward=365, simulations=1000, confidence_levels=[0.95, 0.68]
):
"""
Project future Bitcoin prices using Monte Carlo simulation with era-specific adjustments.
Project future Bitcoin prices with simplified calibration.
"""
# Calculate market maturity score
maturity_score = calculate_market_maturity_score(df)
final_maturity = maturity_score.iloc[-1]
df = df.copy()
df["Log_Price"] = np.log(df["Close"])
df["Log_Return"] = df["Log_Price"].diff()
# Define market eras and their characteristics
min_date = df["Date"].min()
max_date = df["Date"].max()
era_adjustments = {
"early": {
"start_date": pd.Timestamp("2013-01-01"),
"end_date": pd.Timestamp("2017-12-10"), # Futures introduction
"volatility_scale": 1.5, # Balanced for post-2013 early market
"trend_scale": 0.85, # Moderately conservative trends
"skew_scale": 1.2, # Moderate trend following
},
"transition": {
"start_date": pd.Timestamp("2017-12-10"),
"end_date": pd.Timestamp("2020-01-01"),
"volatility_scale": 1.2, # Slightly elevated uncertainty
"trend_scale": 0.95, # Near-normal trends
"skew_scale": 1.05, # Light trend following
},
"mature": {
"start_date": pd.Timestamp("2020-01-01"),
"end_date": pd.Timestamp("2100-01-01"),
"volatility_scale": 0.9, # Slightly reduced volatility for mature market
"trend_scale": 1.0, # Base case
"skew_scale": 0.95, # Slight reduction in trend following
},
}
# Determine which era we're in
current_era = None
for era, params in era_adjustments.items():
if min_date >= params["start_date"] and min_date < params["end_date"]:
current_era = era
break
if current_era is None:
current_era = "mature" # Default to mature era if no match
# Get era-specific adjustment factors
vol_scale = era_adjustments[current_era]["volatility_scale"]
trend_scale = era_adjustments[current_era]["trend_scale"]
skew_scale = era_adjustments[current_era]["skew_scale"]
# Get cycle trends and position
cycle_trends = analyze_trends(df)
# Get current cycle position
halving_dates = get_halving_dates()
current_date = df["Date"].max()
cycle_position = get_cycle_position(current_date, halving_dates)
@@ -396,49 +271,59 @@ def project_prices(
last_price = df["Close"].iloc[-1]
last_date = df["Date"].iloc[-1]
# Generate projection dates
# Generate dates for projection
future_dates = pd.date_range(
start=last_date + timedelta(days=1), periods=days_forward, freq="D"
)
# Calculate expected returns
# Calculate expected returns with cycle boundary handling
future_cycle_days = [
(current_cycle_days + i) % (4 * 365) for i in range(days_forward)
]
expected_returns = []
cycle_trends = analyze_trends(df)
for day in future_cycle_days:
base_trend = cycle_trends.get(day, cycle_trends.mean())
# Get base expected returns
expected_returns = np.array(
[cycle_trends.get(day, cycle_trends.mean()) for day in future_cycle_days]
)
# Add slight mean reversion for extreme values
if abs(base_trend) > 2 * cycle_trends.std():
base_trend *= 0.8 # Dampen extreme trends
# Apply trend adjustments
expected_returns = adjust_trend_expectations(expected_returns, cycle_position)
expected_returns.append(base_trend)
expected_returns = np.array(expected_returns)
# Calculate and adjust volatility
# Calculate base volatility
base_volatility = calculate_volatility(df)
volatility = base_volatility * vol_scale
# Adjust expected returns
adjusted_expected_returns = expected_returns * trend_scale
# 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 for scaling uncertainty over time
projection_adjustments = get_projection_adjustments(days_forward, cycle_position)
# Run Monte Carlo simulation
np.random.seed(42)
simulated_paths = np.zeros((days_forward, simulations))
for sim in range(simulations):
# Calculate skew with era-specific scaling
base_skew = 0.087 * skew_scale
skew = np.sign(adjusted_expected_returns) * base_skew
# Apply era-specific adjustments
drift = expected_returns * era_params["trend_scale"]
vol = base_volatility * era_params["volatility_scale"]
returns = np.random.normal(
loc=adjusted_expected_returns + skew * volatility,
scale=volatility,
size=days_forward,
)
# Scale volatility by projection adjustments
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)
@@ -449,6 +334,11 @@ def project_prices(
results = pd.DataFrame(index=future_dates)
results["Median"] = np.percentile(simulated_paths, 50, axis=1)
# Calculate Expected_Trend using adjusted drift
cumulative_drift = np.cumsum(drift)
results["Expected_Trend"] = last_price * np.exp(cumulative_drift)
# Calculate confidence intervals
for level in confidence_levels:
lower_percentile = (1 - level) * 100 / 2
upper_percentile = 100 - lower_percentile
@@ -460,12 +350,6 @@ def project_prices(
simulated_paths, upper_percentile, axis=1
)
results["Expected_Trend"] = last_price * np.exp(
np.cumsum(adjusted_expected_returns)
)
results["Market_Maturity"] = final_maturity
results["Era"] = current_era
return results
@@ -564,10 +448,6 @@ def create_plots(df, start=None, end=None, project_days=365):
if len(plot_df) == 0:
raise ValueError("No data found for the specified date range")
# Calculate market maturity score
maturity_score = calculate_market_maturity_score(plot_df)
plot_df["Market_Maturity"] = maturity_score
# Generate projections
projections = project_prices(plot_df, days_forward=project_days)
@@ -575,13 +455,13 @@ def create_plots(df, start=None, end=None, project_days=365):
plt.style.use("seaborn-v0_8")
# Create figure with adjusted size for additional subplot
fig = plt.figure(figsize=(15, 20)) # Increased height to accommodate new subplot
fig = plt.figure(figsize=(15, 15)) # Increased height to accommodate new subplot
# Date range for titles
hist_date_range = f" ({plot_df['Date'].min().strftime('%Y-%m-%d')} to {plot_df['Date'].max().strftime('%Y-%m-%d')})"
# 1. Price history and projections (log scale)
ax1 = plt.subplot(5, 1, 1)
ax1 = plt.subplot(4, 1, 1)
# Plot historical prices
ax1.semilogy(plot_df["Date"], plot_df["Close"], "b-", label="Historical Price")
@@ -631,29 +511,8 @@ def create_plots(df, start=None, end=None, project_days=365):
ax1.set_title("Bitcoin Price History and Projections (Log Scale)" + hist_date_range)
ax1.legend(fontsize=8)
# 2. Market Maturity Score
ax2 = plt.subplot(5, 1, 2)
ax2.plot(
plot_df["Date"],
plot_df["Market_Maturity"],
color="purple",
label="Market Maturity Score",
)
ax2.set_title("Market Maturity Score" + hist_date_range)
ax2.set_ylim(0, 1)
ax2.grid(True, alpha=0.3)
ax2.legend()
# Add futures launch annotation
futures_date = pd.Timestamp("2017-12-10")
if futures_date >= plot_df["Date"].min() and futures_date <= plot_df["Date"].max():
ax2.axvline(futures_date, color="red", linestyle="--", alpha=0.5)
ax2.text(
futures_date, 0.95, "Futures\nLaunch", rotation=90, va="top", ha="right"
)
# 3. Rolling volatility
ax3 = plt.subplot(5, 1, 3)
ax3 = plt.subplot(5, 1, 2)
ax3.plot(
plot_df["Date"],
plot_df["Rolling_Volatility_30d"],
@@ -667,7 +526,7 @@ def create_plots(df, start=None, end=None, project_days=365):
ax3.legend()
# 4. Returns distribution
ax4 = plt.subplot(5, 1, 4)
ax4 = plt.subplot(5, 1, 3)
returns_mean = plot_df["Daily_Return"].mean()
returns_std = plot_df["Daily_Return"].std()
filtered_returns = plot_df["Daily_Return"][
@@ -695,7 +554,7 @@ def create_plots(df, start=None, end=None, project_days=365):
)
# 5. Projection ranges
ax5 = plt.subplot(5, 1, 5)
ax5 = plt.subplot(5, 1, 4)
timepoints = np.array(range(30, project_days, 30))
timepoints = timepoints[timepoints <= project_days]
@@ -1159,8 +1018,8 @@ def create_backtest_plot(
def run_projection(df, start):
projections = create_plots(df, start=start, project_days=365 * 4)
#print("\nProjected Prices at Key Points:")
#print(projections.iloc[[29, 89, 179, 364]].round(2)) # 30, 90, 180, 365 days
# print("\nProjected Prices at Key Points:")
# print(projections.iloc[[29, 89, 179, 364]].round(2)) # 30, 90, 180, 365 days
def run_backtest(params, df):