File size: 4,522 Bytes
d5036bb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 |
"""
Market Risk Engine - Usage Examples
Simple examples showing how to use the Market Risk Engine.
"""
import asyncio
from src.core.macroeconomics.markets.risk_engine import (
MarketRiskEngine,
RiskRegime,
RiskEngineConfig,
)
async def example_1_basic_usage():
"""Example 1: Basic risk assessment."""
print("\n=== Example 1: Basic Risk Assessment ===\n")
# Initialize engine
engine = MarketRiskEngine()
# Get current risk score
risk_score = await engine.assess_current_risk()
# Display results
print(f"Global Risk Score: {risk_score.risk_score:.1f}/100")
print(f"Risk Regime: {risk_score.regime.value.upper()}")
# Act based on regime
if risk_score.regime == RiskRegime.RED:
print("⚠️ HIGH RISK: Consider defensive positioning")
elif risk_score.regime == RiskRegime.YELLOW:
print("⚡ MODERATE RISK: Monitor closely")
else:
print("✅ LOW RISK: Favorable environment for risk assets")
async def example_2_category_breakdown():
"""Example 2: Detailed category breakdown."""
print("\n=== Example 2: Category Breakdown ===\n")
engine = MarketRiskEngine()
risk_score = await engine.assess_current_risk()
print("Risk by Category:")
for category, score in risk_score.category_scores.items():
cat_name = category.value.replace('_', ' ').title()
print(f" {cat_name:20s} {score.score:5.1f}/100")
async def example_3_custom_config():
"""Example 3: Custom configuration."""
print("\n=== Example 3: Custom Configuration ===\n")
# Create custom config
config = RiskEngineConfig(
lookback_days=252 # 1 year instead of default 2 years
)
engine = MarketRiskEngine(config=config)
risk_score = await engine.assess_current_risk()
print(f"Risk Score (1-year lookback): {risk_score.risk_score:.1f}/100")
async def example_4_full_report():
"""Example 4: Generate full report."""
print("\n=== Example 4: Full Report ===\n")
engine = MarketRiskEngine()
# Generate comprehensive report
report = await engine.get_full_report()
print(report)
async def example_5_top_risks():
"""Example 5: Identify top risk categories."""
print("\n=== Example 5: Top Risk Categories ===\n")
engine = MarketRiskEngine()
risk_score = await engine.assess_current_risk()
# Get top 3 most stressed categories
top_risks = engine.aggregator.get_top_risks(risk_score, n=3)
print("Top 3 Risk Areas:")
for i, cat_score in enumerate(top_risks, 1):
cat_name = cat_score.category.value.replace('_', ' ').title()
print(f"\n{i}. {cat_name}")
print(f" Score: {cat_score.score:.1f}/100")
print(f" {cat_score.description}")
async def example_6_programmatic_usage():
"""Example 6: Programmatic decision making."""
print("\n=== Example 6: Programmatic Usage ===\n")
engine = MarketRiskEngine()
risk_score = await engine.assess_current_risk()
# Make trading decisions based on risk
if risk_score.risk_score > 70:
print("🔴 Risk Score > 70: Reduce exposure")
print(" Action: Move to defensive positions (TLT, GLD, cash)")
elif risk_score.risk_score > 50:
print("⚡ Risk Score 50-70: Cautious positioning")
print(" Action: Reduce leverage, tighten stops")
else:
print("✅ Risk Score < 50: Favorable for risk assets")
print(" Action: Normal positioning, consider opportunities")
# Check specific categories
credit_score = risk_score.category_scores.get(
next(c for c in risk_score.category_scores.keys()
if c.value == "credit")
)
if credit_score and credit_score.score > 65:
print("\n⚠️ Credit stress detected:")
print(" Consider: Reduce HY exposure, focus on quality")
async def main():
"""Run all examples."""
examples = [
example_1_basic_usage,
example_2_category_breakdown,
example_3_custom_config,
example_4_full_report,
example_5_top_risks,
example_6_programmatic_usage,
]
for example in examples:
try:
await example()
except Exception as e:
print(f"\n✗ Example failed: {e}")
import traceback
traceback.print_exc()
print("\n" + "=" * 70 + "\n")
if __name__ == "__main__":
# Run single example
# asyncio.run(example_1_basic_usage())
# Or run all examples
asyncio.run(main())
|