Create risk_engine.py

risk_engine.py

@@ -0,0 +1,81 @@
+import numpy as np
+from typing import Dict, Any
+from config import (
+    MAX_RISK_PER_TRADE,
+    HIGH_VOLATILITY_THRESHOLD,
+    REDUCED_RISK_FACTOR,
+)
+
+
+def compute_stop_distance(atr: float, multiplier: float = 2.0) -> float:
+    return atr * multiplier
+
+
+def compute_position_size(
+    account_equity: float,
+    entry_price: float,
+    stop_distance: float,
+    risk_fraction: float = MAX_RISK_PER_TRADE,
+) -> float:
+    if stop_distance <= 0 or entry_price <= 0:
+        return 0.0
+    dollar_risk = account_equity * risk_fraction
+    units = dollar_risk / stop_distance
+    notional = units * entry_price
+    return notional
+
+
+def compute_risk_fraction(
+    vol_ratio: float,
+    regime_score: float,
+    base_risk: float = MAX_RISK_PER_TRADE,
+) -> float:
+    risk = base_risk
+
+    if vol_ratio > HIGH_VOLATILITY_THRESHOLD:
+        risk *= REDUCED_RISK_FACTOR
+
+    if regime_score < 0.4:
+        risk *= REDUCED_RISK_FACTOR
+    elif regime_score < 0.6:
+        risk *= 0.75
+
+    return float(np.clip(risk, 0.001, base_risk))
+
+
+def evaluate_risk(
+    df_last_close: float,
+    atr: float,
+    atr_pct: float,
+    regime_score: float,
+    vol_ratio: float,
+    account_equity: float = 10000.0,
+    stop_multiplier: float = 2.0,
+) -> Dict[str, Any]:
+    stop_distance = compute_stop_distance(atr, stop_multiplier)
+    risk_fraction = compute_risk_fraction(vol_ratio, regime_score)
+    position_notional = compute_position_size(
+        account_equity, df_last_close, stop_distance, risk_fraction
+    )
+
+    stop_price_long = df_last_close - stop_distance
+    stop_price_short = df_last_close + stop_distance
+    risk_reward_target = stop_distance * 2.0
+
+    target_long = df_last_close + risk_reward_target
+    target_short = df_last_close - risk_reward_target
+
+    return {
+        "entry_price": df_last_close,
+        "atr": atr,
+        "atr_pct": atr_pct,
+        "stop_distance": stop_distance,
+        "stop_price_long": stop_price_long,
+        "stop_price_short": stop_price_short,
+        "target_long": target_long,
+        "target_short": target_short,
+        "risk_fraction": risk_fraction,
+        "position_notional": position_notional,
+        "vol_ratio": vol_ratio,
+        "regime_score": regime_score,
+    }