cbazza/Place-Order-Trading-Bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add Loss Protection Manager with multi-layer safety system

Browse Source 
- Daily loss limit (2% / $500 max)
- Consecutive loss breaker (3 losses → 2h cooldown)
- Max drawdown circuit breaker (10% threshold)
- News filter with 30min buffer for high-impact events
- Integrated as SCHRITT 0.5 in enhanced_trading_check_wrapper
- Combined lot multiplier with Equity Curve and Reversal Detector

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
cbazza
2026-01-30 17:18:29 +01:00
co-authored by Claude Opus 4.5
parent ce4e961541
commit 4ac27cc7b4
2 changed files with 2794 additions and 565 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TradingBot_V1.6_Adaptive_Complete_CORRECTED.ipynb
+2140 -565
View File
File diff suppressed because it is too large Load Diff
loss_protection_manager.py
+654
View File
@@ -0,0 +1,654 @@
#!/usr/bin/env python3
"""
🛡️ LOSS PROTECTION MANAGER
Umfassender Verlustschutz für den Trading Bot
FEATURES:
1. Daily Loss Limit - Stoppt Trading nach X% Tagesverlust
2. Consecutive Loss Breaker - Pausiert nach X Verlusten in Folge
3. Max Drawdown Circuit Breaker - Hard Stop bei kritischem Drawdown
4. News Filter - Vermeidet Trading bei High-Impact News
VERWENDUNG:
from loss_protection_manager import LossProtectionManager
lpm = LossProtectionManager()
# Vor jedem Trade prüfen:
allowed, reason, multiplier = lpm.check_trading_allowed()
if not allowed:
print(f"Trading blocked: {reason}")
return None
# Nach jedem Trade updaten:
lpm.record_trade(profit=-50.00, symbol="XAUUSD")
"""
import json
import os
import logging
from datetime import datetime, timedelta
from typing import Dict, Tuple, List, Optional
import requests
logger = logging.getLogger(__name__)
class LossProtectionManager:
"""
Comprehensive Loss Protection System
Features:
- Daily loss limit with auto-reset at midnight
- Consecutive loss tracking with cooldown period
- Max drawdown circuit breaker
- Economic calendar news filter
"""
def __init__(self,
# Daily Loss Limit
daily_loss_limit_pct: float = 2.0,
daily_loss_limit_dollars: float = 500.0,
# Consecutive Loss Breaker
max_consecutive_losses: int = 3,
cooldown_minutes: int = 120,
# Max Drawdown Circuit Breaker
max_drawdown_pct: float = 10.0,
drawdown_recovery_pct: float = 5.0,
# News Filter
news_filter_enabled: bool = True,
news_buffer_minutes: int = 30,
block_high_impact: bool = True,
block_medium_impact: bool = False,
# General
data_file: str = "loss_protection_state.json",
account_balance: float = 100000.0):
"""
Args:
daily_loss_limit_pct: Max daily loss as % of account (default: 2%)
daily_loss_limit_dollars: Max daily loss in dollars (default: $500)
max_consecutive_losses: Losses in a row before pause (default: 3)
cooldown_minutes: Pause duration after consecutive losses (default: 2h)
max_drawdown_pct: Circuit breaker threshold (default: 10%)
drawdown_recovery_pct: Recovery needed to resume (default: 5%)
news_filter_enabled: Enable news filtering (default: True)
news_buffer_minutes: Minutes before/after news to avoid (default: 30)
block_high_impact: Block during high-impact news (default: True)
block_medium_impact: Block during medium-impact news (default: False)
data_file: File for persistent state
account_balance: Account balance for % calculations
"""
# Daily Loss Settings
self.daily_loss_limit_pct = daily_loss_limit_pct
self.daily_loss_limit_dollars = daily_loss_limit_dollars
# Consecutive Loss Settings
self.max_consecutive_losses = max_consecutive_losses
self.cooldown_minutes = cooldown_minutes
# Drawdown Settings
self.max_drawdown_pct = max_drawdown_pct
self.drawdown_recovery_pct = drawdown_recovery_pct
# News Settings
self.news_filter_enabled = news_filter_enabled
self.news_buffer_minutes = news_buffer_minutes
self.block_high_impact = block_high_impact
self.block_medium_impact = block_medium_impact
# General
self.data_file = data_file
self.account_balance = account_balance
# State
self.state = self._load_state()
self._check_daily_reset()
# News cache
self.news_cache = []
self.news_cache_time = None
self._log_initialization()
# ==========================================
# MAIN CHECK METHOD
# ==========================================
def check_trading_allowed(self, mt5_account_info=None) -> Tuple[bool, str, float]:
"""
Hauptprüfung ob Trading erlaubt ist
Args:
mt5_account_info: Optional MT5 account info for live balance
Returns:
(allowed, reason, lot_multiplier)
- allowed: True wenn Trading erlaubt
- reason: Erklärung
- lot_multiplier: 1.0 = normal, 0.5 = reduziert, 0.0 = blockiert
"""
# Update account balance if provided
if mt5_account_info:
self.account_balance = mt5_account_info.balance
# Check 1: Daily Loss Limit
daily_allowed, daily_reason = self._check_daily_loss_limit()
if not daily_allowed:
return False, f"🛑 DAILY LIMIT: {daily_reason}", 0.0
# Check 2: Consecutive Losses
consec_allowed, consec_reason, consec_mult = self._check_consecutive_losses()
if not consec_allowed:
return False, f"🛑 CONSECUTIVE LOSSES: {consec_reason}", 0.0
# Check 3: Max Drawdown Circuit Breaker
dd_allowed, dd_reason = self._check_drawdown_circuit_breaker()
if not dd_allowed:
return False, f"🛑 CIRCUIT BREAKER: {dd_reason}", 0.0
# Check 4: News Filter
news_allowed, news_reason, news_mult = self._check_news_filter()
if not news_allowed:
return False, f"🛑 NEWS FILTER: {news_reason}", 0.0
# All checks passed - calculate final multiplier
final_multiplier = min(consec_mult, news_mult)
# Build status message
status_parts = []
if self.state['daily_loss'] != 0:
status_parts.append(f"Daily: ${self.state['daily_loss']:.2f}")
if self.state['consecutive_losses'] > 0:
status_parts.append(f"Consec: {self.state['consecutive_losses']}")
if status_parts:
reason = f"✅ Trading allowed ({', '.join(status_parts)})"
else:
reason = "✅ All protection checks passed"
if final_multiplier < 1.0:
reason += f" | Lot: {final_multiplier:.0%}"
return True, reason, final_multiplier
# ==========================================
# INDIVIDUAL CHECKS
# ==========================================
def _check_daily_loss_limit(self) -> Tuple[bool, str]:
"""Prüft Daily Loss Limit"""
self._check_daily_reset()
daily_loss = abs(self.state['daily_loss'])
limit_dollars = self.daily_loss_limit_dollars
limit_pct = self.daily_loss_limit_pct
# Calculate % loss
if self.account_balance > 0:
loss_pct = (daily_loss / self.account_balance) * 100
else:
loss_pct = 0
# Check dollar limit
if daily_loss >= limit_dollars:
return False, f"${daily_loss:.2f} loss today (limit: ${limit_dollars:.2f})"
# Check percentage limit
if loss_pct >= limit_pct:
return False, f"{loss_pct:.1f}% loss today (limit: {limit_pct:.1f}%)"
return True, f"${daily_loss:.2f} / ${limit_dollars:.2f} ({loss_pct:.1f}%)"
def _check_consecutive_losses(self) -> Tuple[bool, str, float]:
"""Prüft Consecutive Loss Breaker"""
consec = self.state['consecutive_losses']
cooldown_until = self.state.get('cooldown_until')
# Check if in cooldown
if cooldown_until:
cooldown_time = datetime.fromisoformat(cooldown_until)
if datetime.now() < cooldown_time:
remaining = (cooldown_time - datetime.now()).total_seconds() / 60
return False, f"Cooldown active ({remaining:.0f} min remaining)", 0.0
else:
# Cooldown expired, reset
self.state['cooldown_until'] = None
self.state['consecutive_losses'] = 0
self._save_state()
# Check consecutive losses
if consec >= self.max_consecutive_losses:
# Activate cooldown
cooldown_until = datetime.now() + timedelta(minutes=self.cooldown_minutes)
self.state['cooldown_until'] = cooldown_until.isoformat()
self._save_state()
return False, f"{consec} consecutive losses - {self.cooldown_minutes}min cooldown activated", 0.0
# Calculate multiplier based on streak
if consec == 0:
multiplier = 1.0
elif consec == 1:
multiplier = 0.75 # Reduce after 1 loss
elif consec == 2:
multiplier = 0.5 # Reduce more after 2 losses
else:
multiplier = 0.25 # Minimal size
return True, f"{consec}/{self.max_consecutive_losses} consecutive losses", multiplier
def _check_drawdown_circuit_breaker(self) -> Tuple[bool, str]:
"""Prüft Max Drawdown Circuit Breaker"""
if self.state.get('circuit_breaker_active', False):
# Check if recovered enough
peak = self.state.get('peak_equity', self.account_balance)
current = self.account_balance
recovery_target = peak * (1 - (self.max_drawdown_pct - self.drawdown_recovery_pct) / 100)
if current >= recovery_target:
# Recovered, deactivate circuit breaker
self.state['circuit_breaker_active'] = False
self._save_state()
return True, "Circuit breaker deactivated - recovered"
else:
recovery_needed = recovery_target - current
return False, f"Circuit breaker active - need ${recovery_needed:.2f} recovery"
# Calculate current drawdown
peak = self.state.get('peak_equity', self.account_balance)
if self.account_balance > peak:
self.state['peak_equity'] = self.account_balance
peak = self.account_balance
self._save_state()
if peak > 0:
drawdown_pct = ((peak - self.account_balance) / peak) * 100
else:
drawdown_pct = 0
if drawdown_pct >= self.max_drawdown_pct:
# Activate circuit breaker
self.state['circuit_breaker_active'] = True
self._save_state()
return False, f"{drawdown_pct:.1f}% drawdown - circuit breaker ACTIVATED"
return True, f"Drawdown: {drawdown_pct:.1f}% (limit: {self.max_drawdown_pct:.1f}%)"
def _check_news_filter(self) -> Tuple[bool, str, float]:
"""Prüft News Filter"""
if not self.news_filter_enabled:
return True, "News filter disabled", 1.0
# Get upcoming news
news = self._get_upcoming_news()
if not news:
return True, "No high-impact news nearby", 1.0
# Check for news within buffer period
now = datetime.now()
buffer = timedelta(minutes=self.news_buffer_minutes)
for event in news:
event_time = event.get('time')
if event_time:
if isinstance(event_time, str):
try:
event_time = datetime.fromisoformat(event_time.replace('Z', '+00:00'))
except:
continue
time_diff = abs((event_time - now).total_seconds() / 60)
if time_diff <= self.news_buffer_minutes:
impact = event.get('impact', 'unknown')
title = event.get('title', 'Unknown Event')
if impact == 'high' and self.block_high_impact:
return False, f"High-impact: {title} in {time_diff:.0f}min", 0.0
elif impact == 'medium' and self.block_medium_impact:
return False, f"Medium-impact: {title} in {time_diff:.0f}min", 0.0
elif impact == 'high':
# Don't block but reduce size
return True, f"Caution: {title} in {time_diff:.0f}min", 0.5
return True, "No concerning news", 1.0
def _get_upcoming_news(self) -> List[Dict]:
"""Holt kommende News-Events (mit Caching)"""
# Use cache if recent (15 min)
if self.news_cache_time and (datetime.now() - self.news_cache_time).seconds < 900:
return self.news_cache
# Try to fetch from economic calendar API
try:
# ForexFactory-style calendar (simplified)
# In production, use a proper economic calendar API
self.news_cache = self._fetch_economic_calendar()
self.news_cache_time = datetime.now()
return self.news_cache
except Exception as e:
logger.debug(f"Could not fetch news: {e}")
return []
def _fetch_economic_calendar(self) -> List[Dict]:
"""
Fetcht Economic Calendar Events
In production sollte hier eine echte API verwendet werden:
- ForexFactory API
- Investing.com Calendar
- FXStreet Calendar
- etc.
"""
# Simplified: Return empty list or static high-impact events
# This is a placeholder - implement real API integration as needed
# Example static high-impact events (USD-focused for Gold trading)
static_events = [
# These would normally come from an API
# {"title": "FOMC Rate Decision", "time": "2026-01-30T19:00:00", "impact": "high", "currency": "USD"},
# {"title": "Non-Farm Payrolls", "time": "2026-02-07T13:30:00", "impact": "high", "currency": "USD"},
]
return static_events
# ==========================================
# TRADE RECORDING
# ==========================================
def record_trade(self, profit: float, symbol: str = "XAUUSD"):
"""
Zeichnet einen abgeschlossenen Trade auf
Args:
profit: Gewinn/Verlust des Trades
symbol: Gehandeltes Symbol
"""
self._check_daily_reset()
# Update daily P/L
self.state['daily_loss'] += profit if profit < 0 else 0
self.state['daily_profit'] += profit if profit > 0 else 0
self.state['daily_trades'] += 1
# Update consecutive losses
if profit < 0:
self.state['consecutive_losses'] += 1
self.state['consecutive_wins'] = 0
else:
self.state['consecutive_losses'] = 0
self.state['consecutive_wins'] += 1
# Update peak equity tracking
self.account_balance += profit # Approximate update
if self.account_balance > self.state.get('peak_equity', 0):
self.state['peak_equity'] = self.account_balance
# Log
logger.info(f"📊 Trade recorded: ${profit:+.2f} | Daily: ${self.state['daily_loss']:.2f} | Consec losses: {self.state['consecutive_losses']}")
self._save_state()
def reset_consecutive_losses(self):
"""Setzt Consecutive Loss Counter zurück (z.B. nach manuellem Review)"""
self.state['consecutive_losses'] = 0
self.state['cooldown_until'] = None
self._save_state()
logger.info("🔄 Consecutive losses reset")
def reset_daily_stats(self):
"""Setzt Tagesstatistiken zurück"""
self.state['daily_loss'] = 0
self.state['daily_profit'] = 0
self.state['daily_trades'] = 0
self.state['last_reset_date'] = datetime.now().date().isoformat()
self._save_state()
logger.info("🔄 Daily stats reset")
def deactivate_circuit_breaker(self):
"""Deaktiviert Circuit Breaker manuell (Vorsicht!)"""
self.state['circuit_breaker_active'] = False
self._save_state()
logger.warning("⚠️ Circuit breaker manually deactivated!")
# ==========================================
# STATUS & REPORTING
# ==========================================
def get_status(self) -> Dict:
"""Gibt vollständigen Status zurück"""
self._check_daily_reset()
# Calculate daily P/L percentage
daily_pnl = self.state['daily_profit'] + self.state['daily_loss']
daily_pnl_pct = (daily_pnl / self.account_balance * 100) if self.account_balance > 0 else 0
# Calculate drawdown
peak = self.state.get('peak_equity', self.account_balance)
drawdown_pct = ((peak - self.account_balance) / peak * 100) if peak > 0 else 0
return {
'trading_allowed': self.check_trading_allowed()[0],
# Daily Stats
'daily_loss': self.state['daily_loss'],
'daily_profit': self.state['daily_profit'],
'daily_pnl': daily_pnl,
'daily_pnl_pct': daily_pnl_pct,
'daily_trades': self.state['daily_trades'],
'daily_limit_pct': self.daily_loss_limit_pct,
'daily_limit_dollars': self.daily_loss_limit_dollars,
# Consecutive Losses
'consecutive_losses': self.state['consecutive_losses'],
'consecutive_wins': self.state['consecutive_wins'],
'max_consecutive_losses': self.max_consecutive_losses,
'cooldown_active': self.state.get('cooldown_until') is not None,
'cooldown_until': self.state.get('cooldown_until'),
# Drawdown
'current_drawdown_pct': drawdown_pct,
'max_drawdown_limit': self.max_drawdown_pct,
'circuit_breaker_active': self.state.get('circuit_breaker_active', False),
'peak_equity': peak,
'current_equity': self.account_balance,
# News
'news_filter_enabled': self.news_filter_enabled,
'upcoming_news': self._get_upcoming_news()[:3], # Top 3
}
def get_report(self) -> str:
"""Generiert formatierten Status-Report"""
status = self.get_status()
report = []
report.append("")
report.append("=" * 60)
report.append("🛡️ LOSS PROTECTION STATUS")
report.append("=" * 60)
# Overall Status
if status['trading_allowed']:
report.append(" Status: ✅ TRADING ALLOWED")
else:
report.append(" Status: 🛑 TRADING BLOCKED")
report.append("")
# Daily Loss Section
report.append("📅 DAILY LIMITS:")
daily_pct = abs(status['daily_loss']) / self.account_balance * 100 if self.account_balance > 0 else 0
report.append(f" Loss Today: ${abs(status['daily_loss']):,.2f} ({daily_pct:.1f}%)")
report.append(f" Limit: ${status['daily_limit_dollars']:,.2f} ({status['daily_limit_pct']}%)")
report.append(f" Trades Today: {status['daily_trades']}")
report.append("")
# Consecutive Losses Section
report.append("🔢 CONSECUTIVE LOSSES:")
report.append(f" Current Streak: {status['consecutive_losses']}/{status['max_consecutive_losses']}")
if status['cooldown_active']:
report.append(f" Cooldown Until: {status['cooldown_until']}")
else:
report.append(f" Cooldown: Not active")
report.append("")
# Drawdown Section
report.append("📉 DRAWDOWN CIRCUIT BREAKER:")
report.append(f" Current DD: {status['current_drawdown_pct']:.1f}%")
report.append(f" Limit: {status['max_drawdown_limit']}%")
report.append(f" Peak Equity: ${status['peak_equity']:,.2f}")
report.append(f" Current Equity: ${status['current_equity']:,.2f}")
if status['circuit_breaker_active']:
report.append(f" Circuit Breaker: 🔴 ACTIVE")
else:
report.append(f" Circuit Breaker: ✅ Inactive")
report.append("")
# News Section
report.append("📰 NEWS FILTER:")
report.append(f" Enabled: {'Yes' if status['news_filter_enabled'] else 'No'}")
if status['upcoming_news']:
report.append(f" Upcoming Events: {len(status['upcoming_news'])}")
for event in status['upcoming_news'][:2]:
report.append(f" - {event.get('title', 'Unknown')} ({event.get('impact', '?')})")
else:
report.append(f" Upcoming Events: None in buffer period")
report.append("")
report.append("=" * 60)
return "\n".join(report)
# ==========================================
# HELPER METHODS
# ==========================================
def _check_daily_reset(self):
"""Prüft ob Tagesstatistiken zurückgesetzt werden müssen"""
today = datetime.now().date().isoformat()
last_reset = self.state.get('last_reset_date', '')
if today != last_reset:
logger.info(f"📅 New day detected - resetting daily stats")
self.state['daily_loss'] = 0
self.state['daily_profit'] = 0
self.state['daily_trades'] = 0
self.state['last_reset_date'] = today
self._save_state()
def _load_state(self) -> Dict:
"""Lädt State aus Datei"""
default_state = {
'daily_loss': 0,
'daily_profit': 0,
'daily_trades': 0,
'last_reset_date': datetime.now().date().isoformat(),
'consecutive_losses': 0,
'consecutive_wins': 0,
'cooldown_until': None,
'circuit_breaker_active': False,
'peak_equity': self.account_balance,
}
try:
if os.path.exists(self.data_file):
with open(self.data_file, 'r') as f:
loaded = json.load(f)
# Merge with defaults
for key in default_state:
if key not in loaded:
loaded[key] = default_state[key]
return loaded
except Exception as e:
logger.warning(f"Could not load state: {e}")
return default_state
def _save_state(self):
"""Speichert State in Datei"""
try:
with open(self.data_file, 'w') as f:
json.dump(self.state, f, indent=2)
except Exception as e:
logger.error(f"Could not save state: {e}")
def _log_initialization(self):
"""Loggt Initialisierung"""
logger.info("=" * 60)
logger.info("🛡️ LOSS PROTECTION MANAGER INITIALIZED")
logger.info("=" * 60)
logger.info(f" Daily Loss Limit: {self.daily_loss_limit_pct}% / ${self.daily_loss_limit_dollars}")
logger.info(f" Max Consec. Losses: {self.max_consecutive_losses} (cooldown: {self.cooldown_minutes}min)")
logger.info(f" Max Drawdown: {self.max_drawdown_pct}%")
logger.info(f" News Filter: {'Enabled' if self.news_filter_enabled else 'Disabled'}")
logger.info(f" Account Balance: ${self.account_balance:,.2f}")
logger.info("=" * 60)
# ==========================================
# INTEGRATION HELPER
# ==========================================
def create_loss_protection_check(lpm: LossProtectionManager):
"""
Erstellt eine Check-Funktion für die Bot-Integration
Usage:
loss_protection_check = create_loss_protection_check(lpm)
# In trading wrapper:
allowed, reason, mult = loss_protection_check()
"""
def check(mt5_account_info=None):
return lpm.check_trading_allowed(mt5_account_info)
return check
# ==========================================
# STANDALONE TESTING
# ==========================================
if __name__ == "__main__":
print("=" * 60)
print("🛡️ LOSS PROTECTION MANAGER TEST")
print("=" * 60)
# Create manager
lpm = LossProtectionManager(
daily_loss_limit_pct=2.0,
daily_loss_limit_dollars=500.0,
max_consecutive_losses=3,
cooldown_minutes=120,
max_drawdown_pct=10.0,
news_filter_enabled=True,
account_balance=100000.0
)
# Show initial status
print(lpm.get_report())
# Simulate some trades
print("\n🧪 Simulating trades...")
# Winning trade
lpm.record_trade(150.0, "XAUUSD")
allowed, reason, mult = lpm.check_trading_allowed()
print(f"After win: {reason} | Mult: {mult}")
# Losing trades
for i in range(3):
lpm.record_trade(-100.0, "XAUUSD")
allowed, reason, mult = lpm.check_trading_allowed()
print(f"After loss {i+1}: {reason} | Mult: {mult}")
# Show final status
print(lpm.get_report())