Spaces:
Runtime error
Runtime error
| """Macro & Economic Engine – computes risk indices and alerts.""" | |
| from typing import List, Dict, Optional, Any | |
| from ..common.logging import logger | |
| from ..common.exceptions import DataValidationError | |
| from .schemas import MacroResult, MacroRiskAlert | |
| from .risk import ( | |
| score_inflation, | |
| score_interest_rate, | |
| extract_latest_inflation, | |
| extract_latest_interest_rate, | |
| analyze_liquidity, | |
| analyze_forex, | |
| analyze_gdp_divergence, | |
| analyze_global, | |
| ) | |
| class MacroEngine: | |
| """ | |
| Computes macro risk indices and alerts from economic data. | |
| """ | |
| # Weights for composite macro risk index | |
| WEIGHTS = { | |
| "inflation": 0.25, | |
| "interest_rate": 0.25, | |
| "forex": 0.20, | |
| "gdp_divergence": 0.20, | |
| "global": 0.10, | |
| } | |
| def __init__(self, weights: Optional[Dict[str, float]] = None): | |
| self.weights = weights or self.WEIGHTS.copy() | |
| total = sum(self.weights.values()) | |
| if not 0.99 <= total <= 1.01: | |
| logger.warning(f"Weights sum to {total}, normalizing") | |
| self.weights = {k: v / total for k, v in self.weights.items()} | |
| # ---------------------------------------------------------------------- | |
| # Public method | |
| # ---------------------------------------------------------------------- | |
| def compute( | |
| self, | |
| inflation_data: Optional[List[Dict]], | |
| interest_rates_data: Optional[List[Dict]], | |
| nrb_data: Optional[Dict], | |
| forex_data: Optional[List[Dict]], | |
| gdp_data: Optional[Dict], | |
| global_data: Optional[Dict] = None, | |
| ) -> MacroResult: | |
| """ | |
| Process macro data and return risk scores and alerts. | |
| Args: | |
| inflation_data: list from fetch_inflation | |
| interest_rates_data: list from fetch_interest_rates | |
| nrb_data: dict from fetch_nrb_data | |
| forex_data: list from fetch_forex_rates | |
| gdp_data: dict from fetch_gdp_vs_nepse | |
| global_data: dict from fetch_global_market_data | |
| Returns: | |
| MacroResult with macro_risk_index, liquidity_pressure_score, alerts. | |
| """ | |
| alerts = [] | |
| # 1. Inflation shock risk | |
| inflation_rate = extract_latest_inflation(inflation_data) | |
| inflation_risk = score_inflation(inflation_rate) | |
| if inflation_risk > 70: | |
| alerts.append(MacroRiskAlert( | |
| alert_type="inflation_shock", | |
| message=f"High inflation: {inflation_rate:.1f}%", | |
| severity="high", | |
| value=inflation_rate | |
| )) | |
| # 2. Interest rate tightening risk | |
| interest_rate = extract_latest_interest_rate(interest_rates_data) | |
| rate_risk = score_interest_rate(interest_rate) | |
| if rate_risk > 70: | |
| alerts.append(MacroRiskAlert( | |
| alert_type="rate_tightening", | |
| message=f"High interest rates: {interest_rate:.1f}%", | |
| severity="high", | |
| value=interest_rate | |
| )) | |
| # 3. Liquidity pressure from NRB data | |
| liquidity_score, liquidity_alerts = analyze_liquidity(nrb_data) | |
| for la in liquidity_alerts: | |
| alerts.append(MacroRiskAlert( | |
| alert_type=la["type"], | |
| message=la["message"], | |
| severity=la["severity"] | |
| )) | |
| # 4. Currency pressure (forex) | |
| forex_pressure, forex_alert = analyze_forex(forex_data) | |
| if forex_alert: | |
| alerts.append(MacroRiskAlert( | |
| alert_type=forex_alert["type"], | |
| message=forex_alert["message"], | |
| severity=forex_alert["severity"], | |
| value=forex_alert.get("value") | |
| )) | |
| # 5. Macro divergence (GDP vs NEPSE) | |
| divergence_score, div_alert = analyze_gdp_divergence(gdp_data) | |
| if div_alert: | |
| alerts.append(MacroRiskAlert( | |
| alert_type=div_alert["type"], | |
| message=div_alert["message"], | |
| severity=div_alert["severity"], | |
| value=div_alert.get("value") | |
| )) | |
| # 6. Global market impact | |
| global_risk = analyze_global(global_data) | |
| if global_risk > 70: | |
| alerts.append(MacroRiskAlert( | |
| alert_type="global_market_risk", | |
| message="Global market weakness may impact NEPSE", | |
| severity="warning", | |
| value=global_risk | |
| )) | |
| # Combine into composite macro risk index | |
| macro_risk = ( | |
| inflation_risk * self.weights["inflation"] + | |
| rate_risk * self.weights["interest_rate"] + | |
| forex_pressure * self.weights["forex"] + | |
| divergence_score * self.weights["gdp_divergence"] + | |
| global_risk * self.weights["global"] | |
| ) | |
| macro_risk = min(100, max(0, macro_risk)) | |
| return MacroResult( | |
| macro_risk_index=round(macro_risk, 1), | |
| liquidity_pressure_score=round(liquidity_score, 1), | |
| policy_risk_alerts=alerts, | |
| ) |