File size: 3,586 Bytes
02cdf61 |
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 |
# learning_hub/policy_engine.py
from .schemas import ReflectorOutput
import json
class PolicyEngine:
def __init__(self):
# 1. سياسة قبول الدلتا (من النقطة 5)
self.auto_accept_threshold: float = 0.80
# 2. عتبة المراجعة البشرية (من النقطة 7)
# إذا كانت الخسارة (بالنسبة المئوية) أكبر من هذا الحد، أرسلها للمراجعة
self.human_review_loss_threshold_percent: float = -3.0
# 3. قواعد السلامة (Meta-Policy) (من النقطة 7)
self.admissibility_rules = {
"NO_SHORT_TRADES": "النظام لا يدعم صفقات SHORT.",
"MAX_LOSS_PER_TRADE": "يجب ألا يتجاوز وقف الخسارة الأولي 10%."
}
print("✅ Learning Hub Module: Policy Engine loaded")
def get_delta_acceptance(self, reflector_output: ReflectorOutput, trade_pnl_percent: float) -> (bool, str):
"""
يحدد ما إذا كان يجب قبول "الدلتا" المقترحة تلقائياً أم إرسالها للمراجعة.
(تنفيذ النقطة 5 و 7)
"""
# 1. فحص سياسة المراجعة البشرية (النقطة 7)
if trade_pnl_percent < self.human_review_loss_threshold_percent:
reason = f"Human Review Required: Significant loss ({trade_pnl_percent:.2f}%)"
print(f"⚠️ [PolicyEngine] {reason}")
return False, reason # (approved=False)
# 2. فحص سياسة القبول التلقائي (النقطة 5)
reflector_confidence = reflector_output.confidence
# (تبسيط لسياسة النتيجة - سنعتمد على ثقة المنعكس كبداية)
# delta_score = 0.5 * reflector_confidence + ...
delta_score = reflector_confidence
if delta_score >= self.auto_accept_threshold:
reason = f"Auto-Accepted: High confidence ({delta_score:.2f} >= {self.auto_accept_threshold})"
print(f"✅ [PolicyEngine] {reason}")
return True, reason # (approved=True)
else:
reason = f"Human Review Required: Low confidence ({delta_score:.2f} < {self.auto_accept_threshold})"
print(f"ℹ️ [PolicyEngine] {reason}")
return False, reason # (approved=False)
def check_meta_policy_on_decision(self, decision: dict) -> (bool, str):
"""
يفحص قرار (قبل التنفيذ) مقابل قواعد السلامة العليا (Meta-Policy).
(تنفيذ النقطة 7)
"""
# 1. قاعدة "لا لصفقات SHORT"
if decision.get("trade_type") == "SHORT":
return False, self.admissibility_rules["NO_SHORT_TRADES"]
# 2. قاعدة "الحد الأقصى لوقف الخسارة"
action = decision.get("action")
if action == "BUY":
current_price = decision.get("current_price", 1) # (سيتم تمريره)
stop_loss = decision.get("stop_loss", 0)
if current_price > 0 and stop_loss > 0:
loss_percent = ((current_price - stop_loss) / current_price) * 100
if loss_percent > 10.0:
reason = f"Meta-Policy Violation: Stop loss ({loss_percent:.1f}%) exceeds max (10%)"
return False, reason
# إذا مرت جميع الفحوصات
return True, "Meta-Policy Passed" |