#!/usr/bin/env python3 """ 📈 Equity Curve Trading Module Meta-Strategy: Trade nur wenn Equity über Moving Average KONZEPT: - Trackt Equity-Historie nach jedem Trade - Berechnet Moving Average der Equity - Erlaubt Trading nur wenn Equity >= MA - Reduziert Drawdowns durch automatische Pausen VERWENDUNG: from equity_curve_trading import EquityCurveManager ecm = EquityCurveManager(ma_period=10) # Vor jedem Trade prüfen: if ecm.should_trade(): execute_trade(...) # Nach jedem Trade updaten: ecm.update_equity() """ import json import os from datetime import datetime, timezone, timedelta from typing import List, Dict, Optional, Tuple import logging logger = logging.getLogger(__name__) class EquityCurveManager: """ Equity Curve Trading Manager Features: - Automatisches Equity-Tracking - Konfigurierbarer MA-Zeitraum - Optionaler "Soft Mode" (reduzierte Lots statt Stop) - Persistente Speicherung der Historie - Recovery-Erkennung """ def __init__(self, ma_period: int = 10, min_trades_required: int = 5, soft_mode: bool = True, soft_mode_multiplier: float = 0.5, recovery_buffer_pct: float = 0.5, data_file: str = "equity_curve_history.json"): """ Args: ma_period: Anzahl der Trades für Moving Average (default: 10) min_trades_required: Minimum Trades bevor Filter aktiv wird (default: 5) soft_mode: True = reduzierte Lots, False = komplett stoppen soft_mode_multiplier: Lot-Multiplikator wenn unter MA (default: 0.5 = 50%) recovery_buffer_pct: Prozent über MA für "Recovery" Status (default: 0.5%) data_file: Datei für persistente Speicherung """ self.ma_period = ma_period self.min_trades = min_trades_required self.soft_mode = soft_mode self.soft_multiplier = soft_mode_multiplier self.recovery_buffer = recovery_buffer_pct / 100 self.data_file = data_file # Equity Historie laden oder initialisieren self.equity_history: List[Dict] = [] self._load_history() # Status self.current_status = "ACTIVE" # ACTIVE, PAUSED, RECOVERY self.trades_while_paused = 0 logger.info("=" * 60) logger.info("📈 EQUITY CURVE TRADING INITIALIZED") logger.info("=" * 60) logger.info(f" MA Period: {ma_period} trades") logger.info(f" Min Trades: {min_trades_required}") logger.info(f" Mode: {'Soft (reduced lots)' if soft_mode else 'Hard (full stop)'}") if soft_mode: logger.info(f" Soft Multiplier: {soft_mode_multiplier:.0%}") logger.info(f" Recovery Buffer: {recovery_buffer_pct}%") logger.info(f" History File: {data_file}") logger.info(f" Loaded Trades: {len(self.equity_history)}") logger.info("=" * 60) # ========================================== # CORE METHODS # ========================================== def should_trade(self, mt5_account_info=None) -> Tuple[bool, str, float]: """ Prüft ob Trading erlaubt ist basierend auf Equity Curve Args: mt5_account_info: Optional MT5 account info object Returns: (should_trade, reason, lot_multiplier) - should_trade: True wenn traden erlaubt - reason: Erklärung - lot_multiplier: 1.0 = normal, 0.5 = reduziert, etc. """ # Nicht genug Historie if len(self.equity_history) < self.min_trades: return True, f"Warmup: {len(self.equity_history)}/{self.min_trades} trades", 1.0 # Aktuelle Equity holen current_equity = self._get_current_equity(mt5_account_info) if current_equity is None: logger.error("Could not get equity from MT5 — blocking trade as safe default") return False, "Equity unavailable — trade blocked for safety", 0.0 # MA berechnen ma_equity = self._calculate_ma() # Status bestimmen equity_vs_ma_pct = ((current_equity - ma_equity) / ma_equity) * 100 if current_equity >= ma_equity * (1 + self.recovery_buffer): # Deutlich über MA = ACTIVE self.current_status = "ACTIVE" self.trades_while_paused = 0 return True, f"✅ Equity ${current_equity:,.2f} > MA ${ma_equity:,.2f} (+{equity_vs_ma_pct:.1f}%)", 1.0 elif current_equity >= ma_equity: # Knapp über MA = RECOVERY (vorsichtig) self.current_status = "RECOVERY" if self.soft_mode: return True, f"🔄 Recovery: ${current_equity:,.2f} ≈ MA ${ma_equity:,.2f} ({equity_vs_ma_pct:+.1f}%)", 0.75 else: return True, f"🔄 Recovery: ${current_equity:,.2f} ≈ MA ${ma_equity:,.2f}", 1.0 else: # Unter MA = PAUSED oder SOFT self.current_status = "PAUSED" self.trades_while_paused += 1 if self.soft_mode: return True, f"⚠️ Soft Mode: ${current_equity:,.2f} < MA ${ma_equity:,.2f} ({equity_vs_ma_pct:.1f}%)", self.soft_multiplier else: return False, f"⛔ PAUSED: ${current_equity:,.2f} < MA ${ma_equity:,.2f} ({equity_vs_ma_pct:.1f}%)", 0.0 def update_equity(self, mt5_account_info=None, trade_result: Optional[Dict] = None): """ Updated Equity-Historie nach einem Trade Args: mt5_account_info: Optional MT5 account info trade_result: Optional dict mit Trade-Details """ current_equity = self._get_current_equity(mt5_account_info) if current_equity is None: logger.warning("Could not get equity for update") return entry = { "timestamp": datetime.now(timezone.utc).isoformat(), "equity": current_equity, "trade_count": len(self.equity_history) + 1 } if trade_result: entry["trade_profit"] = trade_result.get("profit", 0) entry["trade_symbol"] = trade_result.get("symbol", "UNKNOWN") self.equity_history.append(entry) self._save_history() # Log status ma = self._calculate_ma() if len(self.equity_history) >= self.min_trades else None if ma: diff_pct = ((current_equity - ma) / ma) * 100 status_emoji = "✅" if current_equity >= ma else "⚠️" logger.info(f"📈 Equity Update: ${current_equity:,.2f} | MA: ${ma:,.2f} | {status_emoji} {diff_pct:+.1f}%") else: logger.info(f"📈 Equity Update: ${current_equity:,.2f} | Warmup: {len(self.equity_history)}/{self.min_trades}") def get_status(self, mt5_account_info=None) -> Dict: """ Gibt detaillierten Status zurück Returns: Dict mit allen relevanten Informationen """ current_equity = self._get_current_equity(mt5_account_info) ma = self._calculate_ma() if len(self.equity_history) >= self.min_trades else None status = { "current_equity": current_equity, "ma_equity": ma, "ma_period": self.ma_period, "total_trades": len(self.equity_history), "min_trades_required": self.min_trades, "warmup_complete": len(self.equity_history) >= self.min_trades, "status": self.current_status, "soft_mode": self.soft_mode, "soft_multiplier": self.soft_multiplier if self.soft_mode else None } if current_equity and ma: status["equity_vs_ma_pct"] = ((current_equity - ma) / ma) * 100 status["equity_above_ma"] = current_equity >= ma return status def get_report(self, mt5_account_info=None) -> str: """ Generiert einen formatierten Status-Report """ status = self.get_status(mt5_account_info) report = [] report.append("") report.append("=" * 60) report.append("📈 EQUITY CURVE TRADING STATUS") report.append("=" * 60) if status["current_equity"]: report.append(f" Current Equity: ${status['current_equity']:,.2f}") if status["ma_equity"]: report.append(f" MA ({self.ma_period} trades): ${status['ma_equity']:,.2f}") diff = status.get("equity_vs_ma_pct", 0) if status.get("equity_above_ma"): report.append(f" Status: ✅ ABOVE MA (+{diff:.1f}%)") else: report.append(f" Status: ⚠️ BELOW MA ({diff:.1f}%)") else: report.append(f" Status: 🔄 Warmup ({status['total_trades']}/{status['min_trades_required']} trades)") report.append("") report.append(f" Trading Status: {status['status']}") report.append(f" Mode: {'Soft' if status['soft_mode'] else 'Hard'}") if status['soft_mode'] and status['status'] == 'PAUSED': report.append(f" Lot Multiplier: {status['soft_multiplier']:.0%}") report.append("") report.append(f" Total Trades: {status['total_trades']}") report.append("=" * 60) return "\n".join(report) # ========================================== # HELPER METHODS # ========================================== def _get_current_equity(self, mt5_account_info=None) -> Optional[float]: """Holt aktuelle Equity von MT5 oder übergebenem Object""" if mt5_account_info: return mt5_account_info.equity try: import MetaTrader5 as mt5 account = mt5.account_info() if account: return account.equity except Exception as e: logger.debug(f"Could not get MT5 equity: {e}") return None def _calculate_ma(self) -> float: """Berechnet Moving Average der letzten N Equity-Werte""" if len(self.equity_history) < self.ma_period: # Nutze alle verfügbaren wenn nicht genug recent = self.equity_history else: recent = self.equity_history[-self.ma_period:] equities = [entry["equity"] for entry in recent] return sum(equities) / len(equities) if equities else 0 def _load_history(self): """Lädt Equity-Historie aus Datei""" try: if os.path.exists(self.data_file): with open(self.data_file, 'r') as f: self.equity_history = json.load(f) logger.info(f"📂 Loaded {len(self.equity_history)} equity records") except Exception as e: logger.warning(f"Could not load equity history: {e}") self.equity_history = [] def _save_history(self): """Speichert Equity-Historie in Datei""" try: with open(self.data_file, 'w') as f: json.dump(self.equity_history, f, indent=2) except Exception as e: logger.error(f"Could not save equity history: {e}") def reset_history(self): """Setzt Historie zurück (Vorsicht!)""" self.equity_history = [] self._save_history() logger.warning("⚠️ Equity history has been reset!") def add_initial_equity(self, equity: float): """ Fügt initiale Equity hinzu (für Warmup) Nützlich wenn du mit bestehendem Konto startest """ base_time = datetime.now(timezone.utc) for i in range(self.min_trades): ts = (base_time - timedelta(minutes=self.min_trades - i)).isoformat() self.equity_history.append({ "timestamp": ts, "equity": equity, "trade_count": i + 1, "note": "Initial warmup entry" }) self._save_history() logger.info(f"📈 Added {self.min_trades} initial equity entries at ${equity:,.2f}") # ========================================== # STANDALONE USAGE # ========================================== if __name__ == "__main__": # Demo print("📈 Equity Curve Trading Demo") print("=" * 50) ecm = EquityCurveManager( ma_period=5, min_trades_required=3, soft_mode=True, soft_mode_multiplier=0.5 ) # Simulate some trades test_equities = [10000, 10200, 10150, 9900, 9700, 9500, 9600, 9800, 10000, 10300] print("\nSimulating trades:") for i, eq in enumerate(test_equities): # Fake the history ecm.equity_history.append({ "timestamp": datetime.now().isoformat(), "equity": eq, "trade_count": i + 1 }) # Check if should trade should, reason, mult = ecm.should_trade() print(f"Trade {i+1}: Equity ${eq:,} | {reason} | Lot mult: {mult}") print("\n" + ecm.get_report())