| """ |
| 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") |
|
|
| |
| engine = MarketRiskEngine() |
|
|
| |
| risk_score = await engine.assess_current_risk() |
|
|
| |
| print(f"Global Risk Score: {risk_score.risk_score:.1f}/100") |
| print(f"Risk Regime: {risk_score.regime.value.upper()}") |
|
|
| |
| 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") |
|
|
| |
| config = RiskEngineConfig( |
| lookback_days=252 |
| ) |
|
|
| 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() |
|
|
| |
| 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() |
|
|
| |
| 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() |
|
|
| |
| 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") |
|
|
| |
| 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__": |
| |
| |
|
|
| |
| asyncio.run(main()) |
|
|
