- Replace non-existent Python examples with actual files - Fix all placeholder yourusername URLs to ThotDjehuty - Remove references to non-existent optimal_control.md theory doc - Update examples to reference: hmm_regime_detection.py, parallel_de_benchmark.py, polaroid_optimizr_integration.py, timeseries_integration.py
31 KiB
31 KiB
In [1]:
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
from optimizr import (
HMM,
mcmc_sample,
grid_search,
mutual_information,
shannon_entropy
)
np.random.seed(42)
sns.set_style('whitegrid')
print("✓ All modules loaded successfully!")✓ All modules loaded successfully!
In [2]:
def generate_realistic_market_data(n_days=730, start_price=50000):
"""
Generate realistic crypto market data with regime switching.
Returns:
DataFrame with prices, returns, and true regimes
"""
# Define 3 market regimes
regimes = {
0: {'name': 'Bull', 'mu': 0.0015, 'sigma': 0.025, 'color': 'green'}, # ~40% annual
1: {'name': 'Bear', 'mu': -0.0010, 'sigma': 0.040, 'color': 'red'}, # -25% annual, high vol
2: {'name': 'Neutral', 'mu': 0.0002, 'sigma': 0.020, 'color': 'gray'} # ~5% annual
}
# Transition matrix (regimes tend to persist)
transition_matrix = np.array([
[0.95, 0.03, 0.02], # Bull tends to stay bull
[0.05, 0.90, 0.05], # Bear tends to stay bear
[0.15, 0.10, 0.75] # Neutral is transition state
])
# Generate state sequence
states = [0] # Start in bull market
for _ in range(n_days - 1):
current = states[-1]
next_state = np.random.choice(3, p=transition_matrix[current])
states.append(next_state)
states = np.array(states)
# Generate returns with regime-dependent parameters
returns = np.zeros(n_days)
for t in range(n_days):
regime = states[t]
mu = regimes[regime]['mu']
sigma = regimes[regime]['sigma']
# Add GARCH-like volatility clustering
if t > 0:
vol_shock = 0.3 * abs(returns[t-1]) / sigma
sigma *= (1 + vol_shock)
returns[t] = np.random.normal(mu, sigma)
# Generate prices
prices = start_price * np.exp(np.cumsum(returns))
# Create DataFrame
dates = [datetime(2023, 1, 1) + timedelta(days=i) for i in range(n_days)]
df = pd.DataFrame({
'date': dates,
'price': prices,
'return': returns,
'true_regime': states,
'regime_name': [regimes[s]['name'] for s in states]
})
return df, regimes
# Generate data
df_btc, regime_info = generate_realistic_market_data(n_days=730, start_price=50000)
print(f"Generated {len(df_btc)} days of market data")
print(f"\nPrice range: ${df_btc['price'].min():,.0f} - ${df_btc['price'].max():,.0f}")
print(f"\nRegime distribution:")
print(df_btc['regime_name'].value_counts())
print(f"\nReturn statistics:")
print(f"Mean: {df_btc['return'].mean():.4f} ({df_btc['return'].mean()*252:.2%} annual)")
print(f"Std: {df_btc['return'].std():.4f} ({df_btc['return'].std()*np.sqrt(252):.2%} annual)")
print(f"Sharpe: {df_btc['return'].mean() / df_btc['return'].std() * np.sqrt(252):.2f}")Generated 730 days of market data Price range: $20,745 - $82,080 Regime distribution: regime_name Bull 424 Bear 214 Neutral 92 Name: count, dtype: int64 Return statistics: Mean: 0.0004 (9.14% annual) Std: 0.0400 (63.46% annual) Sharpe: 0.14
In [ ]:
fig, axes = plt.subplots(3, 1, figsize=(15, 10))
# Plot 1: Price with regime colors
for regime_id, info in regime_info.items():
mask = df_btc['true_regime'] == regime_id
axes[0].scatter(df_btc.loc[mask, 'date'], df_btc.loc[mask, 'price'],
c=info['color'], label=info['name'], alpha=0.6, s=10)
axes[0].set_ylabel('Price ($)', fontsize=12)
axes[0].set_title('BTC Price with True Market Regimes', fontsize=14, fontweight='bold')
axes[0].legend(loc='upper left')
axes[0].grid(alpha=0.3)
# Plot 2: Daily returns
axes[1].plot(df_btc['date'], df_btc['return'], linewidth=0.8, alpha=0.7, color='blue')
axes[1].axhline(0, color='black', linestyle='--', alpha=0.3)
axes[1].fill_between(df_btc['date'], 0, df_btc['return'],
where=df_btc['return']>0, alpha=0.3, color='green', label='Positive')
axes[1].fill_between(df_btc['date'], 0, df_btc['return'],
where=df_btc['return']<0, alpha=0.3, color='red', label='Negative')
axes[1].set_ylabel('Daily Return', fontsize=12)
axes[1].set_title('Daily Returns', fontsize=14, fontweight='bold')
axes[1].legend()
axes[1].grid(alpha=0.3)
# Plot 3: Return distribution
axes[2].hist(df_btc['return'], bins=50, alpha=0.7, color='skyblue', edgecolor='black', density=True)
x_range = np.linspace(df_btc['return'].min(), df_btc['return'].max(), 100)
from scipy import stats
axes[2].plot(x_range, stats.norm.pdf(x_range, df_btc['return'].mean(), df_btc['return'].std()),
'r-', linewidth=2, label='Normal fit')
axes[2].set_xlabel('Daily Return', fontsize=12)
axes[2].set_ylabel('Density', fontsize=12)
axes[2].set_title('Return Distribution (Note: Fat Tails)', fontsize=14, fontweight='bold')
axes[2].legend()
axes[2].grid(alpha=0.3)
plt.tight_layout()
plt.show()In [4]:
# Fit HMM to detect regimes
print("Fitting HMM to detect market regimes...")
hmm = HMM(n_states=3)
hmm.fit(df_btc['return'].values, n_iterations=100, tolerance=1e-6)
print("\n✓ HMM fitted successfully!")
print(f"\nLearned Transition Matrix:")
print(pd.DataFrame(hmm.transition_matrix_,
columns=['State 0', 'State 1', 'State 2'],
index=['State 0', 'State 1', 'State 2']).round(3))
print(f"\nEmission Parameters:")
print(pd.DataFrame({
'Mean Return': hmm.emission_means_,
'Volatility': hmm.emission_stds_,
'Annual Return': hmm.emission_means_ * 252,
'Annual Vol': hmm.emission_stds_ * np.sqrt(252)
}, index=['State 0', 'State 1', 'State 2']).round(4))Fitting HMM to detect market regimes...
✓ HMM fitted successfully!
Learned Transition Matrix:
State 0 State 1 State 2
State 0 0.339 0.247 0.414
State 1 0.252 0.441 0.306
State 2 0.459 0.215 0.326
Emission Parameters:
Mean Return Volatility Annual Return Annual Vol
State 0 -0.0370 0.0297 -9.3169 0.4716
State 1 0.0019 0.0117 0.4693 0.1850
State 2 0.0369 0.0281 9.3038 0.4462
In [ ]:
# Predict states
returns_array = np.asarray(df_btc['return'].values, dtype=np.float64)
predicted_states = hmm.predict(returns_array)
df_btc['predicted_regime'] = predicted_states
# Map states to regime names based on mean returns
state_means = hmm.emission_means_
state_mapping = {}
sorted_states = np.argsort(state_means)[::-1] # Highest mean first
regime_names_sorted = ['Bull', 'Neutral', 'Bear']
for i, state in enumerate(sorted_states):
state_mapping[state] = regime_names_sorted[i]
df_btc['predicted_regime_name'] = df_btc['predicted_regime'].map(state_mapping)
print("\nPredicted regime distribution:")
print(df_btc['predicted_regime_name'].value_counts())In [ ]:
fig, axes = plt.subplots(2, 1, figsize=(15, 10), sharex=True)
# Plot 1: True regimes
regime_colors = {'Bull': 'green', 'Bear': 'red', 'Neutral': 'gray'}
for regime_name in ['Bull', 'Bear', 'Neutral']:
mask = df_btc['regime_name'] == regime_name
axes[0].scatter(df_btc.loc[mask, 'date'], df_btc.loc[mask, 'price'],
c=regime_colors[regime_name], label=regime_name, alpha=0.6, s=15)
axes[0].set_ylabel('Price ($)', fontsize=12)
axes[0].set_title('True Market Regimes', fontsize=14, fontweight='bold')
axes[0].legend()
axes[0].grid(alpha=0.3)
# Plot 2: Predicted regimes
for regime_name in ['Bull', 'Bear', 'Neutral']:
mask = df_btc['predicted_regime_name'] == regime_name
axes[1].scatter(df_btc.loc[mask, 'date'], df_btc.loc[mask, 'price'],
c=regime_colors[regime_name], label=f'Predicted {regime_name}', alpha=0.6, s=15)
axes[1].set_xlabel('Date', fontsize=12)
axes[1].set_ylabel('Price ($)', fontsize=12)
axes[1].set_title('HMM-Detected Regimes', fontsize=14, fontweight='bold')
axes[1].legend()
axes[1].grid(alpha=0.3)
plt.tight_layout()
plt.show()
# Calculate accuracy
from itertools import permutations
def calculate_best_accuracy(true_labels, pred_labels):
true_regime_map = {'Bull': 0, 'Bear': 1, 'Neutral': 2}
true_numeric = df_btc['regime_name'].map(true_regime_map).values
best_acc = 0
for perm in permutations([0, 1, 2]):
mapped = np.array([perm[s] for s in pred_labels])
acc = np.mean(mapped == true_numeric)
best_acc = max(best_acc, acc)
return best_acc
accuracy = calculate_best_accuracy(df_btc['regime_name'], predicted_states)
print(f"\n✓ Regime detection accuracy: {accuracy:.2%}")In [ ]:
def log_posterior_returns(params, data):
"""
Log posterior for return distribution parameters.
"""
mu, sigma = params
if sigma <= 0:
return -np.inf
# Log-likelihood
residuals = (data - mu) / sigma
log_lik = -0.5 * len(data) * np.log(2 * np.pi)
log_lik -= len(data) * np.log(sigma)
log_lik -= 0.5 * np.sum(residuals**2)
# Log-prior for mu ~ N(0, 0.1^2)
log_prior_mu = -0.5 * (mu / 0.1)**2
# Log-prior for sigma ~ LogNormal(log(0.02), 0.5)
log_prior_sigma = -0.5 * ((np.log(sigma) - np.log(0.02)) / 0.5)**2 - np.log(sigma)
return log_lik + log_prior_mu + log_prior_sigma
# Run MCMC for Bull regime
bull_returns = df_btc[df_btc['predicted_regime_name'] == 'Bull']['return'].values
print(f"Running MCMC for Bull regime ({len(bull_returns)} observations)...")
mcmc_samples, acceptance_rate = mcmc_sample(
log_likelihood_fn=log_posterior_returns,
data=bull_returns,
initial_params=[0.001, 0.02],
param_bounds=[(-0.01, 0.01), (0.001, 0.1)],
proposal_std=[0.0002, 0.002],
n_samples=15000,
burn_in=2000
)
print(f"\n✓ MCMC completed! Acceptance rate: {acceptance_rate:.2%}")
print(f"\nPosterior estimates (Bull regime):")
print(f"μ: {mcmc_samples[:, 0].mean():.5f} ± {mcmc_samples[:, 0].std():.5f}")
print(f"σ: {mcmc_samples[:, 1].mean():.5f} ± {mcmc_samples[:, 1].std():.5f}")
print(f"\nAnnualized:")
print(f"Return: {mcmc_samples[:, 0].mean() * 252:.2%} ± {mcmc_samples[:, 0].std() * 252:.2%}")
print(f"Vol: {mcmc_samples[:, 1].mean() * np.sqrt(252):.2%} ± {mcmc_samples[:, 1].std() * np.sqrt(252):.2%}")In [ ]:
fig, axes = plt.subplots(2, 2, figsize=(14, 10))
# Trace plots
axes[0, 0].plot(mcmc_samples[:, 0] * 252, linewidth=0.5, alpha=0.7)
axes[0, 0].set_ylabel('Annual Return', fontsize=11)
axes[0, 0].set_title('Trace: μ (Bull Regime)', fontsize=13, fontweight='bold')
axes[0, 0].grid(alpha=0.3)
axes[0, 1].plot(mcmc_samples[:, 1] * np.sqrt(252), linewidth=0.5, alpha=0.7, color='orange')
axes[0, 1].set_ylabel('Annual Volatility', fontsize=11)
axes[0, 1].set_title('Trace: σ (Bull Regime)', fontsize=13, fontweight='bold')
axes[0, 1].grid(alpha=0.3)
# Posterior distributions
axes[1, 0].hist(mcmc_samples[:, 0] * 252, bins=50, density=True, alpha=0.7, color='skyblue', edgecolor='black')
axes[1, 0].axvline((mcmc_samples[:, 0] * 252).mean(), color='red', linestyle='--', linewidth=2, label='Mean')
axes[1, 0].set_xlabel('Annual Return', fontsize=11)
axes[1, 0].set_ylabel('Density', fontsize=11)
axes[1, 0].set_title('Posterior: μ', fontsize=13, fontweight='bold')
axes[1, 0].legend()
axes[1, 0].grid(alpha=0.3)
axes[1, 1].hist(mcmc_samples[:, 1] * np.sqrt(252), bins=50, density=True, alpha=0.7, color='orange', edgecolor='black')
axes[1, 1].axvline((mcmc_samples[:, 1] * np.sqrt(252)).mean(), color='red', linestyle='--', linewidth=2, label='Mean')
axes[1, 1].set_xlabel('Annual Volatility', fontsize=11)
axes[1, 1].set_ylabel('Density', fontsize=11)
axes[1, 1].set_title('Posterior: σ', fontsize=13, fontweight='bold')
axes[1, 1].legend()
axes[1, 1].grid(alpha=0.3)
plt.tight_layout()
plt.show()In [ ]:
# Generate correlated second asset
np.random.seed(42)
eth_returns = 0.7 * df_btc['return'].values + 0.3 * np.random.randn(len(df_btc)) * 0.03
eth_returns += 0.0005 # Slight outperformance
# Calculate statistics
returns_matrix = np.column_stack([df_btc['return'].values, eth_returns])
mean_returns = returns_matrix.mean(axis=0)
cov_matrix = np.cov(returns_matrix.T)
print("Asset Statistics:")
print(f"BTC: Return={mean_returns[0]*252:.2%}, Vol={np.sqrt(cov_matrix[0,0]*252):.2%}")
print(f"ETH: Return={mean_returns[1]*252:.2%}, Vol={np.sqrt(cov_matrix[1,1]*252):.2%}")
print(f"\nCorrelation: {cov_matrix[0,1] / (np.sqrt(cov_matrix[0,0]) * np.sqrt(cov_matrix[1,1])):.3f}")
def portfolio_sharpe(weights, mean_ret, cov_mat, rf=0.0):
"""
Calculate negative Sharpe ratio (for minimization).
"""
w = np.array(weights)
# Ensure weights sum to 1
if abs(w.sum() - 1.0) > 0.01:
return -1e10 # Penalty
port_return = w @ mean_ret
port_vol = np.sqrt(w @ cov_mat @ w)
if port_vol < 1e-10:
return -1e10
sharpe = (port_return - rf) / port_vol
return -sharpe # Negative because grid_search maximizes
# Run grid search (weight_btc, weight_eth)
# We'll search over weight_btc, and set weight_eth = 1 - weight_btc
print("\nRunning Grid Search for optimal portfolio...")
def portfolio_objective_1d(params):
w_btc = params[0]
weights = [w_btc, 1 - w_btc]
return portfolio_sharpe(weights, mean_returns, cov_matrix)
result = grid_search(
objective_fn=portfolio_objective_1d,
bounds=[(0.0, 1.0)], # BTC weight from 0 to 1
n_points=100
)
optimal_w_btc = result.x[0]
optimal_w_eth = 1 - optimal_w_btc
optimal_sharpe = -result.fun # Convert back to positive
print(f"\n✓ Grid Search completed!")
print(f"\nOptimal Portfolio:")
print(f"BTC weight: {optimal_w_btc:.1%}")
print(f"ETH weight: {optimal_w_eth:.1%}")
print(f"\nExpected Sharpe Ratio: {optimal_sharpe * np.sqrt(252):.3f}")
optimal_return = optimal_w_btc * mean_returns[0] + optimal_w_eth * mean_returns[1]
optimal_vol = np.sqrt(np.array([optimal_w_btc, optimal_w_eth]) @ cov_matrix @ np.array([optimal_w_btc, optimal_w_eth]))
print(f"Expected Return: {optimal_return * 252:.2%}")
print(f"Expected Volatility: {optimal_vol * np.sqrt(252):.2%}")In [ ]:
# Calculate efficient frontier
weights_range = np.linspace(0, 1, 100)
port_returns = []
port_vols = []
port_sharpes = []
for w_btc in weights_range:
w = np.array([w_btc, 1 - w_btc])
ret = w @ mean_returns * 252
vol = np.sqrt(w @ cov_matrix @ w) * np.sqrt(252)
sharpe = ret / vol if vol > 0 else 0
port_returns.append(ret)
port_vols.append(vol)
port_sharpes.append(sharpe)
fig, axes = plt.subplots(1, 2, figsize=(15, 6))
# Plot 1: Efficient frontier
scatter = axes[0].scatter(port_vols, port_returns, c=port_sharpes, cmap='RdYlGn', s=50, alpha=0.6)
axes[0].scatter([optimal_vol * np.sqrt(252)], [optimal_return * 252],
c='red', s=300, marker='*', edgecolors='black', linewidths=2,
label='Optimal Portfolio', zorder=5)
# Mark individual assets
axes[0].scatter([np.sqrt(cov_matrix[0,0]*252)], [mean_returns[0]*252],
c='blue', s=200, marker='D', edgecolors='black', linewidths=2,
label='BTC Only', zorder=5)
axes[0].scatter([np.sqrt(cov_matrix[1,1]*252)], [mean_returns[1]*252],
c='purple', s=200, marker='D', edgecolors='black', linewidths=2,
label='ETH Only', zorder=5)
plt.colorbar(scatter, ax=axes[0], label='Sharpe Ratio')
axes[0].set_xlabel('Volatility (Annual)', fontsize=12)
axes[0].set_ylabel('Return (Annual)', fontsize=12)
axes[0].set_title('Efficient Frontier', fontsize=14, fontweight='bold')
axes[0].legend()
axes[0].grid(alpha=0.3)
# Plot 2: Sharpe ratio vs BTC weight
axes[1].plot(weights_range * 100, port_sharpes, linewidth=2, color='green')
axes[1].axvline(optimal_w_btc * 100, color='red', linestyle='--', linewidth=2,
label=f'Optimal: {optimal_w_btc:.1%} BTC')
axes[1].fill_between(weights_range * 100, 0, port_sharpes, alpha=0.3, color='green')
axes[1].set_xlabel('BTC Weight (%)', fontsize=12)
axes[1].set_ylabel('Sharpe Ratio', fontsize=12)
axes[1].set_title('Sharpe Ratio vs Portfolio Allocation', fontsize=14, fontweight='bold')
axes[1].legend()
axes[1].grid(alpha=0.3)
plt.tight_layout()
plt.show()In [ ]:
# Calculate entropy of returns (discretized)
btc_entropy = shannon_entropy(df_btc['return'].values)
eth_entropy = shannon_entropy(eth_returns)
print("Shannon Entropy (uncertainty):")
print(f"BTC: {btc_entropy:.4f} bits")
print(f"ETH: {eth_entropy:.4f} bits")
# Calculate mutual information
mi_btc_eth = mutual_information(df_btc['return'].values, eth_returns)
print(f"\nMutual Information (dependency):")
print(f"I(BTC; ETH): {mi_btc_eth:.4f} bits")
print(f"\nNormalized MI (correlation-like): {mi_btc_eth / min(btc_entropy, eth_entropy):.4f}")
# Compare to Pearson correlation
pearson_corr = np.corrcoef(df_btc['return'].values, eth_returns)[0, 1]
print(f"Pearson correlation: {pearson_corr:.4f}")
print(f"\n→ MI captures non-linear dependencies that correlation misses!")In [ ]:
# Calculate rolling MI
window = 90 # 90-day window
rolling_mi = []
rolling_corr = []
dates_rolling = []
for i in range(window, len(df_btc)):
btc_window = df_btc['return'].values[i-window:i]
eth_window = eth_returns[i-window:i]
mi = mutual_information(btc_window, eth_window)
corr = np.corrcoef(btc_window, eth_window)[0, 1]
rolling_mi.append(mi)
rolling_corr.append(corr)
dates_rolling.append(df_btc['date'].iloc[i])
# Visualize
fig, axes = plt.subplots(3, 1, figsize=(15, 12), sharex=True)
# Plot 1: Prices
ax1_twin = axes[0].twinx()
axes[0].plot(df_btc['date'], df_btc['price'], label='BTC', color='orange', linewidth=2)
eth_price = 3000 * np.exp(np.cumsum(eth_returns))
ax1_twin.plot(df_btc['date'], eth_price, label='ETH (synthetic)', color='purple', linewidth=2, alpha=0.7)
axes[0].set_ylabel('BTC Price ($)', fontsize=12, color='orange')
ax1_twin.set_ylabel('ETH Price ($)', fontsize=12, color='purple')
axes[0].set_title('Asset Prices', fontsize=14, fontweight='bold')
axes[0].grid(alpha=0.3)
# Plot 2: Rolling correlation
axes[1].plot(dates_rolling, rolling_corr, color='blue', linewidth=2)
axes[1].axhline(pearson_corr, color='red', linestyle='--', linewidth=2, label='Overall correlation')
axes[1].fill_between(dates_rolling, 0, rolling_corr, alpha=0.3, color='blue')
axes[1].set_ylabel('Correlation', fontsize=12)
axes[1].set_title(f'Rolling {window}-Day Correlation', fontsize=14, fontweight='bold')
axes[1].legend()
axes[1].grid(alpha=0.3)
# Plot 3: Rolling MI
axes[2].plot(dates_rolling, rolling_mi, color='green', linewidth=2)
axes[2].axhline(mi_btc_eth, color='red', linestyle='--', linewidth=2, label='Overall MI')
axes[2].fill_between(dates_rolling, 0, rolling_mi, alpha=0.3, color='green')
axes[2].set_xlabel('Date', fontsize=12)
axes[2].set_ylabel('Mutual Information (bits)', fontsize=12)
axes[2].set_title(f'Rolling {window}-Day Mutual Information', fontsize=14, fontweight='bold')
axes[2].legend()
axes[2].grid(alpha=0.3)
plt.tight_layout()
plt.show()
print(f"\n✓ Dependency analysis reveals time-varying relationship structure!")
print(f"MI standard deviation: {np.std(rolling_mi):.4f} bits")
print(f"Correlation standard deviation: {np.std(rolling_corr):.4f}")