def map_to_known_liquidity_regimes():
"""
Define known Fed policy regimes based on historical record
"""
# Format: (start_date, end_date, regime_type, description)
known_regimes = [
# =====================================
# Pre-Crisis Era (1990-2007)
# =====================================
('1990-01', '1992-12', 'Neutral', 'Post-1990 Recession, Gradual Easing'),
('1993-01', '1994-01', 'High', 'Accommodative Policy, Economic Expansion'),
('1994-02', '1995-12', 'Tight', 'Fed Tightening Cycle (3.0% → 6.0%)'),
('1996-01', '1998-12', 'Neutral', 'Goldilocks Economy'),
('1999-01', '2000-05', 'Tight', 'Dot-com Bubble, Y2K Tightening'),
('2000-06', '2003-06', 'High', 'Dot-com Crash Response, Aggressive Easing'),
('2003-07', '2004-05', 'High', 'Ultra-Low Rates (1%), Housing Boom'),
('2004-06', '2006-06', 'Tight', 'Fed Hiking Cycle (1% → 5.25%)'),
('2006-07', '2007-07', 'Neutral', 'Pause at Peak, Pre-Crisis'),
# =====================================
# Financial Crisis & QE Era (2007-2015)
# =====================================
('2007-08', '2008-08', 'Neutral', 'Initial Crisis Response, Rate Cuts Begin'),
('2008-09', '2010-03', 'High', 'QE1: Emergency Response ($1.7T)'),
('2010-04', '2010-10', 'Neutral', 'QE1 Taper, Brief Pause'),
('2010-11', '2011-06', 'High', 'QE2: $600B Purchase Program'),
('2011-07', '2012-08', 'Neutral', 'Operation Twist, Pre-QE3'),
('2012-09', '2014-10', 'High', 'QE3: $85B/month, then taper'),
('2014-11', '2015-11', 'Neutral', 'Post-QE3, ZIRP Maintained'),
# =====================================
# Normalization Attempt (2015-2019)
# =====================================
('2015-12', '2018-12', 'Tight', 'Fed Hiking + QT (0.25% → 2.50%)'),
('2019-01', '2019-07', 'Neutral', 'Pause, Economic Slowdown'),
('2019-08', '2019-12', 'High', 'Insurance Cuts + Repo Crisis Response'),
# =====================================
# COVID Era (2020-2021)
# =====================================
('2020-01', '2020-02', 'Neutral', 'Pre-COVID'),
('2020-03', '2021-10', 'High', 'COVID QE: Unlimited purchases, ZIRP'),
('2021-11', '2022-02', 'High', 'Taper Announcement but still easy'),
# =====================================
# Modern Tightening (2022-2025)
# =====================================
('2022-03', '2023-06', 'Tight', 'Aggressive Hikes (0% → 5.25%)'),
('2023-07', '2024-08', 'Tight', 'Higher for Longer, QT Active'),
('2024-09', '2025-01', 'Neutral', 'Easing Cycle Begins'),
]
# Convert to DataFrame
regime_df = pd.DataFrame(known_regimes, columns=['start', 'end', 'regime', 'description'])
regime_df['start'] = pd.to_datetime(regime_df['start'])
regime_df['end'] = pd.to_datetime(regime_df['end'])
return regime_df
def validate_hmm_against_known_regimes(regimes_df, known_regimes_df):
"""
Compare HMM-detected regimes to known Fed policy periods
"""
print(f"\n{'='*70}")
print("REGIME VALIDATION: HMM vs Known Fed Policy")
print(f"{'='*70}")
# Expand known regimes to monthly observations
known_monthly = []
for _, row in known_regimes_df.iterrows():
date_range = pd.date_range(row['start'], row['end'], freq='M')
for date in date_range:
known_monthly.append({
'date': date,
'known_regime': row['regime'],
'period_description': row['description']
})
known_df = pd.DataFrame(known_monthly).set_index('date')
# Merge with HMM results
comparison = regimes_df.join(known_df, how='inner')
comparison = comparison.dropna(subset=['known_regime', 'state_label'])
print(f"\n📊 Overlap period: {comparison.index[0].strftime('%Y-%m')} to {comparison.index[-1].strftime('%Y-%m')}")
print(f" Total months: {len(comparison)}")
# =====================================
# Confusion Matrix
# =====================================
print(f"\n{'='*70}")
print("Confusion Matrix: HMM vs Known Regimes")
print(f"{'='*70}")
from sklearn.metrics import confusion_matrix, accuracy_score, classification_report
# Create confusion matrix
cm = confusion_matrix(comparison['known_regime'], comparison['state_label'],
labels=['Tight', 'Neutral', 'High'])
cm_df = pd.DataFrame(cm,
index=['Known_Tight', 'Known_Neutral', 'Known_High'],
columns=['HMM_Tight', 'HMM_Neutral', 'HMM_High'])
print("\n", cm_df)
# Calculate percentages
cm_pct = cm_df.div(cm_df.sum(axis=1), axis=0) * 100
print("\nPercentages (row-wise):")
print(cm_pct.round(1))
# Overall accuracy
accuracy = accuracy_score(comparison['known_regime'], comparison['state_label'])
print(f"\n📈 Overall Accuracy: {accuracy:.1%}")
# =====================================
# Regime-by-Regime Analysis
# =====================================
print(f"\n{'='*70}")
print("Detailed Regime Concordance")
print(f"{'='*70}")
for known_regime in ['Tight', 'Neutral', 'High']:
mask = comparison['known_regime'] == known_regime
subset = comparison[mask]
print(f"\n{known_regime.upper()} REGIME ({mask.sum()} months):")
# How HMM classified these periods
hmm_dist = subset['state_label'].value_counts(normalize=True) * 100
for hmm_regime, pct in hmm_dist.sort_index().items():
status = "✅" if hmm_regime == known_regime else "⚠️"
print(f" {status} Classified as {hmm_regime:8s}: {pct:5.1f}%")
# =====================================
# Key Historical Periods Check
# =====================================
print(f"\n{'='*70}")
print("Key Historical Periods Validation")
print(f"{'='*70}")
key_periods = [
('2008-09', '2009-12', 'High', 'QE1 (Financial Crisis)'),
('2010-11', '2011-06', 'High', 'QE2'),
('2012-09', '2014-10', 'High', 'QE3'),
('2015-12', '2018-12', 'Tight', 'Fed Hiking + QT'),
('2020-03', '2021-06', 'High', 'COVID QE (Unlimited)'),
('2022-03', '2023-12', 'Tight', 'Aggressive Rate Hikes'),
]
for start, end, expected, description in key_periods:
try:
mask = (comparison.index >= start) & (comparison.index <= end)
subset = comparison[mask]
if len(subset) == 0:
continue
# Modal HMM regime
hmm_mode = subset['state_label'].mode()[0]
hmm_pct = (subset['state_label'] == hmm_mode).sum() / len(subset) * 100
# Check if matches expectation
match = "✅" if hmm_mode == expected else "❌"
print(f"\n{match} {description}")
print(f" Period: {start} to {end}")
print(f" Expected: {expected}")
print(f" HMM Mode: {hmm_mode} ({hmm_pct:.0f}% of period)")
# Show distribution
dist = subset['state_label'].value_counts()
print(f" Distribution: ", end="")
for regime, count in dist.items():
print(f"{regime}={count}, ", end="")
print()
except Exception as e:
print(f"\n⚠️ {description}: Data not available")
# =====================================
# Mismatches Analysis
# =====================================
print(f"\n{'='*70}")
print("Significant Mismatches (Known ≠ HMM)")
print(f"{'='*70}")
mismatches = comparison[comparison['known_regime'] != comparison['state_label']]
if len(mismatches) > 0:
# Find consecutive mismatch periods
mismatch_periods = []
current_start = None
for i, (date, row) in enumerate(mismatches.iterrows()):
if current_start is None:
current_start = date
current_known = row['known_regime']
current_hmm = row['state_label']
# Check if next row continues the mismatch pattern
is_last = i == len(mismatches) - 1
if is_last or (not is_last and mismatches.index[i+1] != date + pd.DateOffset(months=1)):
mismatch_periods.append({
'start': current_start,
'end': date,
'known': current_known,
'hmm': current_hmm,
'months': (date.year - current_start.year) * 12 + date.month - current_start.month + 1
})
current_start = None
# Show significant mismatches (>= 3 months)
significant = [p for p in mismatch_periods if p['months'] >= 3]
if significant:
print(f"\nFound {len(significant)} significant mismatch periods (≥3 months):\n")
for p in significant:
print(f" {p['start'].strftime('%Y-%m')} to {p['end'].strftime('%Y-%m')} ({p['months']} months)")
print(f" Known: {p['known']:8s} | HMM: {p['hmm']:8s}")
# Get description for this period
desc_mask = (known_regimes_df['start'] <= p['start']) & (known_regimes_df['end'] >= p['end'])
if desc_mask.any():
desc = known_regimes_df[desc_mask]['description'].iloc[0]
print(f" Context: {desc}")
print()
else:
print("\n✅ No significant consecutive mismatches (all discrepancies < 3 months)")
else:
print("\n🎉 Perfect match! HMM exactly matches known regimes.")
# =====================================
# Summary Statistics
# =====================================
print(f"\n{'='*70}")
print("SUMMARY STATISTICS")
print(f"{'='*70}")
print(f"\n📊 Agreement Rates by Regime:")
for regime in ['Tight', 'Neutral', 'High']:
mask = comparison['known_regime'] == regime
if mask.sum() > 0:
agreement = (comparison[mask]['state_label'] == regime).sum() / mask.sum() * 100
print(f" {regime:8s}: {agreement:5.1f}% correct")
print(f"\n📊 Overall Performance:")
print(f" Total Accuracy: {accuracy:.1%}")
print(f" Mismatch Rate: {(1-accuracy):.1%}")
print(f" Consecutive Mismatches: {len([p for p in mismatch_periods if p['months'] >= 3])}")
# Interpretation
print(f"\n{'='*70}")
print("INTERPRETATION")
print(f"{'='*70}")
if accuracy >= 0.80:
print("\n✅ EXCELLENT: HMM successfully captures Fed policy regimes")
print(" The model shows strong alignment with known monetary policy history.")
elif accuracy >= 0.65:
print("\n✔️ GOOD: HMM generally aligns with Fed policy regimes")
print(" Some discrepancies exist but the model captures major policy shifts.")
elif accuracy >= 0.50:
print("\n⚠️ MODERATE: HMM shows partial alignment with Fed policy")
print(" Consider adjusting liquidity index composition or HMM parameters.")
else:
print("\n❌ POOR: HMM does not align well with Fed policy regimes")
print(" Major revisions needed to liquidity index or regime detection.")
return comparison, cm_df, accuracy
# Run validation
known_regimes_df = map_to_known_liquidity_regimes()
comparison_df, confusion_matrix, accuracy = validate_hmm_against_known_regimes(
regimes_df,
known_regimes_df
)