cbazzaandClaude Sonnet 4.6
338ed1188c
session_filter_patch.py: - Fix mutable default arguments (config=None + internal assignment) - Read confidence threshold from config (95) instead of hardcoded 60 - Read debug flag from config instead of hardcoding True - Rename datetime parameter to avoid shadowing the module (_datetime import) - Clamp optimal_interval to max 59 to avoid % modulo issues - Cache now = _datetime.now() to avoid double call advanced_position_management.py: - mt -> mt5 alias (21 replacements) - should_update_trailing_stop: fetch symbol_info.point once, reuse for both checks - close_partial_position: fetch mt5.symbol_info_tick once instead of twice - check_and_update_positions: add mt5.terminal_info() guard loss_protection_manager.py: - Fix critical bug: .seconds -> .total_seconds() in news cache check (.seconds resets at 1h boundary, causing stale cache to appear fresh) - _fetch_economic_calendar: activate via news_filter_simple integration, document that it was previously a no-op - record_trade: document approximate balance tracking limitation Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
585 lines
21 KiB
Python
585 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
🎯 Advanced Position Management Module
|
|
Performance Optimization Features:
|
|
1. Adaptive Position Sizing
|
|
2. Trailing Stop-Loss
|
|
3. Partial Take Profit
|
|
"""
|
|
|
|
import MetaTrader5 as mt5
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Tuple, Optional, Dict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ==========================================
|
|
# 1. ADAPTIVE POSITION SIZING
|
|
# ==========================================
|
|
|
|
class AdaptivePositionSizer:
|
|
"""
|
|
Passt Position Size basierend auf Signal Confidence an
|
|
|
|
Bessere Signals → Größere Positionen
|
|
Schwächere Signals → Kleinere Positionen
|
|
"""
|
|
|
|
def __init__(self,
|
|
base_risk: float = 0.02,
|
|
high_confidence_threshold: float = 80.0,
|
|
medium_confidence_threshold: float = 70.0,
|
|
high_multiplier: float = 1.5,
|
|
medium_multiplier: float = 1.0,
|
|
low_multiplier: float = 0.5):
|
|
"""
|
|
Args:
|
|
base_risk: Basis-Risk pro Trade (default 1%)
|
|
high_confidence_threshold: Ab diesem Wert gilt Signal als "high confidence"
|
|
medium_confidence_threshold: Ab diesem Wert gilt Signal als "medium confidence"
|
|
high_multiplier: Risk-Multiplikator für high confidence (1.5x = 1.5%)
|
|
medium_multiplier: Risk-Multiplikator für medium confidence (1.0x = 1.0%)
|
|
low_multiplier: Risk-Multiplikator für low confidence (0.5x = 0.5%)
|
|
"""
|
|
self.base_risk = base_risk
|
|
self.high_threshold = high_confidence_threshold
|
|
self.medium_threshold = medium_confidence_threshold
|
|
self.high_mult = high_multiplier
|
|
self.medium_mult = medium_multiplier
|
|
self.low_mult = low_multiplier
|
|
|
|
def calculate_risk_for_confidence(self, confidence: float) -> float:
|
|
"""
|
|
Berechnet angepasstes Risk basierend auf Confidence
|
|
|
|
Args:
|
|
confidence: Signal Confidence (0-100)
|
|
|
|
Returns:
|
|
Angepasstes Risk (z.B. 0.015 für 1.5%)
|
|
"""
|
|
if confidence >= self.high_threshold:
|
|
multiplier = self.high_mult
|
|
category = "HIGH"
|
|
elif confidence >= self.medium_threshold:
|
|
multiplier = self.medium_mult
|
|
category = "MEDIUM"
|
|
else:
|
|
multiplier = self.low_mult
|
|
category = "LOW"
|
|
|
|
adjusted_risk = self.base_risk * multiplier
|
|
|
|
logger.info(f"📊 Adaptive Position Sizing:")
|
|
logger.info(f" Confidence: {confidence:.1f}% ({category})")
|
|
logger.info(f" Base Risk: {self.base_risk*100:.1f}%")
|
|
logger.info(f" Multiplier: {multiplier}x")
|
|
logger.info(f" Adjusted Risk: {adjusted_risk*100:.1f}%")
|
|
|
|
return adjusted_risk
|
|
|
|
def calculate_position_size(self,
|
|
confidence: float,
|
|
balance: float,
|
|
stop_loss_distance: float,
|
|
symbol: str = "XAUUSD") -> float:
|
|
"""
|
|
Berechnet Position Size mit adaptivem Risk
|
|
|
|
Args:
|
|
confidence: Signal Confidence
|
|
balance: Account Balance
|
|
stop_loss_distance: Distanz zum Stop Loss in Pips
|
|
symbol: Trading Symbol
|
|
|
|
Returns:
|
|
Lot Size (Volume)
|
|
"""
|
|
# Adaptive Risk
|
|
adjusted_risk = self.calculate_risk_for_confidence(confidence)
|
|
risk_amount = balance * adjusted_risk
|
|
|
|
# Symbol Info
|
|
symbol_info = mt5.symbol_info(symbol)
|
|
if not symbol_info:
|
|
logger.error(f"Symbol info not available for {symbol}")
|
|
return 0.10 # Minimum
|
|
|
|
# Pip Value berechnen
|
|
point = symbol_info.point
|
|
tick_value = symbol_info.trade_tick_value
|
|
tick_size = symbol_info.trade_tick_size
|
|
pip_value = (tick_value / tick_size) * point
|
|
|
|
# Volume berechnen
|
|
volume = risk_amount / (stop_loss_distance * pip_value)
|
|
|
|
# Auf erlaubte Schritte runden
|
|
volume_min = max(symbol_info.volume_min, 0.10) # Min: Broker-Min oder 0.10
|
|
volume_max = min(symbol_info.volume_max, 0.20) # Max: Broker-Max oder 0.20
|
|
volume_step = symbol_info.volume_step
|
|
|
|
volume = round(volume / volume_step) * volume_step
|
|
volume = max(volume_min, min(volume_max, volume))
|
|
|
|
logger.info(f"💰 Position Size: {volume:.2f} lots")
|
|
logger.info(f" Risk Amount: ${risk_amount:.2f}")
|
|
logger.info(f" SL Distance: {stop_loss_distance:.2f} pips")
|
|
|
|
return volume
|
|
|
|
|
|
# ==========================================
|
|
# 2. TRAILING STOP-LOSS
|
|
# ==========================================
|
|
|
|
class TrailingStopManager:
|
|
"""
|
|
Verwaltet Trailing Stop-Loss für laufende Positionen
|
|
|
|
Bewegt Stop-Loss mit Profit mit:
|
|
- Break-Even bei 50% des Weges zu TP
|
|
- Lock 50% Profit bei 75% des Weges zu TP
|
|
"""
|
|
|
|
def __init__(self,
|
|
breakeven_trigger_pct: float = 0.5,
|
|
profit_lock_trigger_pct: float = 0.75,
|
|
profit_lock_amount_pct: float = 0.5,
|
|
min_distance_points: int = 100):
|
|
"""
|
|
Args:
|
|
breakeven_trigger_pct: Bei wie viel % des TP-Wegs auf Break-Even
|
|
profit_lock_trigger_pct: Bei wie viel % des TP-Wegs Profit locken
|
|
profit_lock_amount_pct: Wie viel % vom Profit locken
|
|
min_distance_points: Minimum Distanz in Points (Anti-Stop-Hunting)
|
|
"""
|
|
self.breakeven_trigger = breakeven_trigger_pct
|
|
self.profit_lock_trigger = profit_lock_trigger_pct
|
|
self.profit_lock_amount = profit_lock_amount_pct
|
|
self.min_distance = min_distance_points
|
|
|
|
def should_update_trailing_stop(self, position) -> Tuple[bool, Optional[float], str]:
|
|
"""
|
|
Prüft ob Trailing Stop aktualisiert werden sollte
|
|
|
|
Args:
|
|
position: MT5 Position Object
|
|
|
|
Returns:
|
|
(should_update, new_sl_price, reason)
|
|
"""
|
|
try:
|
|
# Position Info
|
|
ticket = position.ticket
|
|
position_type = position.type # 0=BUY, 1=SELL
|
|
entry_price = position.price_open
|
|
current_sl = position.sl
|
|
tp = position.tp
|
|
|
|
# Current Price
|
|
symbol_info = mt5.symbol_info_tick(position.symbol)
|
|
if not symbol_info:
|
|
return False, None, "No symbol info"
|
|
|
|
current_price = symbol_info.bid if position_type == 0 else symbol_info.ask
|
|
|
|
# TP Distance
|
|
if position_type == 0: # BUY
|
|
tp_distance = tp - entry_price
|
|
current_distance = current_price - entry_price
|
|
else: # SELL
|
|
tp_distance = entry_price - tp
|
|
current_distance = entry_price - current_price
|
|
|
|
if tp_distance <= 0:
|
|
return False, None, "Invalid TP distance"
|
|
|
|
# Progress to TP
|
|
progress_pct = current_distance / tp_distance
|
|
|
|
# Fetch symbol point once for all distance checks below
|
|
sym_point = mt5.symbol_info(position.symbol).point
|
|
|
|
# Check Break-Even Trigger
|
|
if progress_pct >= self.breakeven_trigger:
|
|
new_sl = entry_price
|
|
|
|
if position_type == 0: # BUY
|
|
sl_distance_points = (current_price - new_sl) / sym_point
|
|
else: # SELL
|
|
sl_distance_points = (new_sl - current_price) / sym_point
|
|
|
|
if sl_distance_points < self.min_distance:
|
|
return False, None, f"Distance too small: {sl_distance_points:.0f} points"
|
|
|
|
if position_type == 0: # BUY
|
|
if current_sl > 0 and new_sl <= current_sl:
|
|
return False, None, "Would move SL backwards"
|
|
else: # SELL
|
|
if current_sl > 0 and new_sl >= current_sl:
|
|
return False, None, "Would move SL backwards"
|
|
|
|
return True, new_sl, f"Break-Even at {progress_pct*100:.1f}% progress"
|
|
|
|
# Check Profit Lock Trigger
|
|
if progress_pct >= self.profit_lock_trigger:
|
|
if position_type == 0: # BUY
|
|
locked_profit = tp_distance * self.profit_lock_amount
|
|
new_sl = entry_price + locked_profit
|
|
else: # SELL
|
|
locked_profit = tp_distance * self.profit_lock_amount
|
|
new_sl = entry_price - locked_profit
|
|
|
|
if position_type == 0: # BUY
|
|
sl_distance_points = (current_price - new_sl) / sym_point
|
|
else: # SELL
|
|
sl_distance_points = (new_sl - current_price) / sym_point
|
|
|
|
if sl_distance_points < self.min_distance:
|
|
return False, None, f"Distance too small: {sl_distance_points:.0f} points"
|
|
|
|
# Don't move SL backwards
|
|
if position_type == 0: # BUY
|
|
if current_sl > 0 and new_sl <= current_sl:
|
|
return False, None, "Would move SL backwards"
|
|
else: # SELL
|
|
if current_sl > 0 and new_sl >= current_sl:
|
|
return False, None, "Would move SL backwards"
|
|
|
|
return True, new_sl, f"Locking {self.profit_lock_amount*100:.0f}% profit at {progress_pct*100:.1f}% progress"
|
|
|
|
return False, None, f"Progress {progress_pct*100:.1f}% < trigger {self.breakeven_trigger*100:.0f}%"
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking trailing stop: {e}")
|
|
return False, None, str(e)
|
|
|
|
def update_stop_loss(self, position, new_sl: float) -> bool:
|
|
"""
|
|
Aktualisiert Stop-Loss für Position
|
|
|
|
Args:
|
|
position: MT5 Position
|
|
new_sl: Neuer Stop-Loss Preis
|
|
|
|
Returns:
|
|
Success
|
|
"""
|
|
try:
|
|
request = {
|
|
"action": mt5.TRADE_ACTION_SLTP,
|
|
"position": position.ticket,
|
|
"symbol": position.symbol,
|
|
"sl": new_sl,
|
|
"tp": position.tp,
|
|
"magic": 234000,
|
|
"comment": "Trailing Stop"
|
|
}
|
|
|
|
result = mt5.order_send(request)
|
|
|
|
if result.retcode == mt5.TRADE_RETCODE_DONE:
|
|
logger.info(f"✅ Trailing Stop updated for #{position.ticket}")
|
|
logger.info(f" Old SL: {position.sl:.5f}")
|
|
logger.info(f" New SL: {new_sl:.5f}")
|
|
return True
|
|
else:
|
|
logger.error(f"❌ Failed to update trailing stop: {result.comment}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error updating stop loss: {e}")
|
|
return False
|
|
|
|
|
|
# ==========================================
|
|
# 3. PARTIAL TAKE PROFIT
|
|
# ==========================================
|
|
|
|
class PartialTakeProfitManager:
|
|
"""
|
|
Verwaltet Partial Take Profit
|
|
|
|
Schließt Teil der Position bei TP1, lässt Rest laufen bis TP2
|
|
"""
|
|
|
|
def __init__(self,
|
|
tp1_risk_ratio: float = 1.5,
|
|
tp2_risk_ratio: float = 2.5,
|
|
partial_close_pct: float = 0.5):
|
|
"""
|
|
Args:
|
|
tp1_risk_ratio: TP1 bei diesem Risk-Reward (1.5 = 1.5x Risk)
|
|
tp2_risk_ratio: TP2 bei diesem Risk-Reward (2.5 = 2.5x Risk)
|
|
partial_close_pct: Wie viel % bei TP1 schließen (0.5 = 50%)
|
|
"""
|
|
self.tp1_ratio = tp1_risk_ratio
|
|
self.tp2_ratio = tp2_risk_ratio
|
|
self.partial_pct = partial_close_pct
|
|
|
|
def calculate_partial_tp_levels(self,
|
|
entry_price: float,
|
|
sl_price: float,
|
|
position_type: int) -> Tuple[float, float]:
|
|
"""
|
|
Berechnet TP1 und TP2 Levels
|
|
|
|
Args:
|
|
entry_price: Entry Preis
|
|
sl_price: Stop Loss Preis
|
|
position_type: 0=BUY, 1=SELL
|
|
|
|
Returns:
|
|
(tp1_price, tp2_price)
|
|
"""
|
|
if position_type == 0: # BUY
|
|
risk = entry_price - sl_price
|
|
tp1 = entry_price + (risk * self.tp1_ratio)
|
|
tp2 = entry_price + (risk * self.tp2_ratio)
|
|
else: # SELL
|
|
risk = sl_price - entry_price
|
|
tp1 = entry_price - (risk * self.tp1_ratio)
|
|
tp2 = entry_price - (risk * self.tp2_ratio)
|
|
|
|
logger.info(f"📊 Partial TP Levels:")
|
|
logger.info(f" Entry: {entry_price:.5f}")
|
|
logger.info(f" SL: {sl_price:.5f}")
|
|
logger.info(f" Risk: {abs(risk):.5f}")
|
|
logger.info(f" TP1 ({self.tp1_ratio}R): {tp1:.5f}")
|
|
logger.info(f" TP2 ({self.tp2_ratio}R): {tp2:.5f}")
|
|
|
|
return tp1, tp2
|
|
|
|
def should_close_partial(self, position, tp1_price: float) -> Tuple[bool, str]:
|
|
"""
|
|
Prüft ob Partial Close ausgeführt werden soll
|
|
|
|
Args:
|
|
position: MT5 Position
|
|
tp1_price: TP1 Preis Level
|
|
|
|
Returns:
|
|
(should_close, reason)
|
|
"""
|
|
try:
|
|
# Current Price
|
|
symbol_info = mt5.symbol_info_tick(position.symbol)
|
|
if not symbol_info:
|
|
return False, "No symbol info"
|
|
|
|
current_price = symbol_info.bid if position.type == 0 else symbol_info.ask
|
|
|
|
# Check if TP1 hit
|
|
if position.type == 0: # BUY
|
|
if current_price >= tp1_price:
|
|
return True, f"TP1 hit: Price {current_price:.5f} >= TP1 {tp1_price:.5f}"
|
|
else: # SELL
|
|
if current_price <= tp1_price:
|
|
return True, f"TP1 hit: Price {current_price:.5f} <= TP1 {tp1_price:.5f}"
|
|
|
|
return False, f"TP1 not reached yet"
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking partial close: {e}")
|
|
return False, str(e)
|
|
|
|
def close_partial_position(self, position, close_pct: float = None) -> bool:
|
|
"""
|
|
Schließt Teil der Position
|
|
|
|
Args:
|
|
position: MT5 Position
|
|
close_pct: Prozent zum Schließen (default: self.partial_pct)
|
|
|
|
Returns:
|
|
Success
|
|
"""
|
|
try:
|
|
if close_pct is None:
|
|
close_pct = self.partial_pct
|
|
|
|
# Calculate volume to close
|
|
close_volume = round(position.volume * close_pct, 2)
|
|
|
|
# Minimum volume check
|
|
symbol_info = mt5.symbol_info(position.symbol)
|
|
if close_volume < symbol_info.volume_min:
|
|
logger.warning(f"Close volume {close_volume} < minimum {symbol_info.volume_min}")
|
|
return False
|
|
|
|
close_type = mt5.ORDER_TYPE_SELL if position.type == 0 else mt5.ORDER_TYPE_BUY
|
|
tick = mt5.symbol_info_tick(position.symbol)
|
|
if not tick:
|
|
logger.error(f"Could not get tick for {position.symbol}")
|
|
return False
|
|
close_price = tick.bid if position.type == 0 else tick.ask
|
|
|
|
request = {
|
|
"action": mt5.TRADE_ACTION_DEAL,
|
|
"position": position.ticket,
|
|
"symbol": position.symbol,
|
|
"volume": close_volume,
|
|
"type": close_type,
|
|
"price": close_price,
|
|
"deviation": 20,
|
|
"magic": 234000,
|
|
"comment": f"Partial TP1 ({close_pct*100:.0f}%)",
|
|
"type_time": mt5.ORDER_TIME_GTC,
|
|
"type_filling": mt5.ORDER_FILLING_IOC,
|
|
}
|
|
|
|
result = mt5.order_send(request)
|
|
|
|
if result.retcode == mt5.TRADE_RETCODE_DONE:
|
|
logger.info(f"✅ Partial close executed for #{position.ticket}")
|
|
logger.info(f" Closed: {close_volume:.2f} lots ({close_pct*100:.0f}%)")
|
|
logger.info(f" Remaining: {position.volume - close_volume:.2f} lots")
|
|
logger.info(f" Profit: ${result.profit:.2f}")
|
|
return True
|
|
else:
|
|
logger.error(f"❌ Partial close failed: {result.comment}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error closing partial position: {e}")
|
|
return False
|
|
|
|
|
|
# ==========================================
|
|
# INTEGRATED MANAGER
|
|
# ==========================================
|
|
|
|
class AdvancedPositionManager:
|
|
"""
|
|
Integrierter Manager für alle Advanced Features
|
|
"""
|
|
|
|
def __init__(self,
|
|
enable_adaptive_sizing: bool = True,
|
|
enable_trailing_stop: bool = True,
|
|
enable_partial_tp: bool = True,
|
|
base_risk: float = 0.02):
|
|
"""
|
|
Args:
|
|
enable_adaptive_sizing: Adaptive Position Sizing aktivieren
|
|
enable_trailing_stop: Trailing Stop aktivieren
|
|
enable_partial_tp: Partial TP aktivieren
|
|
base_risk: Base Risk per Trade (wird an AdaptivePositionSizer übergeben)
|
|
"""
|
|
self.adaptive_sizing = AdaptivePositionSizer(base_risk=base_risk) if enable_adaptive_sizing else None
|
|
self.trailing_stop = TrailingStopManager() if enable_trailing_stop else None
|
|
self.partial_tp = PartialTakeProfitManager() if enable_partial_tp else None
|
|
|
|
# Track partial closes (avoid duplicate partial closes)
|
|
self.partial_closed_positions = set()
|
|
|
|
logger.info("🎯 Advanced Position Manager initialized")
|
|
logger.info(f" Adaptive Sizing: {'✅' if enable_adaptive_sizing else '❌'}")
|
|
logger.info(f" Trailing Stop: {'✅' if enable_trailing_stop else '❌'}")
|
|
logger.info(f" Partial TP: {'✅' if enable_partial_tp else '❌'}")
|
|
|
|
def check_and_update_positions(self, symbol: str = "XAUUSD"):
|
|
"""
|
|
Prüft alle offenen Positionen und aktualisiert Trailing Stops / Partial TPs
|
|
|
|
Args:
|
|
symbol: Symbol zum Checken
|
|
"""
|
|
try:
|
|
if not mt5.terminal_info():
|
|
logger.error("MT5 not initialized — skipping position management")
|
|
return
|
|
|
|
positions = mt5.positions_get(symbol=symbol)
|
|
|
|
if not positions:
|
|
return
|
|
|
|
logger.info(f"\n🔍 Checking {len(positions)} position(s) for {symbol}...")
|
|
|
|
for position in positions:
|
|
# Trailing Stop Check
|
|
if self.trailing_stop:
|
|
should_update, new_sl, reason = self.trailing_stop.should_update_trailing_stop(position)
|
|
|
|
if should_update:
|
|
logger.info(f"📈 Trailing Stop Trigger for #{position.ticket}: {reason}")
|
|
self.trailing_stop.update_stop_loss(position, new_sl)
|
|
else:
|
|
logger.debug(f"⏸️ No trailing stop update: {reason}")
|
|
|
|
# Partial TP Check (only if not already partially closed)
|
|
if self.partial_tp and position.ticket not in self.partial_closed_positions:
|
|
# Calculate TP1 from current position
|
|
tp1, tp2 = self.partial_tp.calculate_partial_tp_levels(
|
|
position.price_open,
|
|
position.sl,
|
|
position.type
|
|
)
|
|
|
|
should_close, reason = self.partial_tp.should_close_partial(position, tp1)
|
|
|
|
if should_close:
|
|
logger.info(f"🎯 Partial TP Trigger for #{position.ticket}: {reason}")
|
|
if self.partial_tp.close_partial_position(position):
|
|
self.partial_closed_positions.add(position.ticket)
|
|
else:
|
|
logger.debug(f"⏸️ No partial close: {reason}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking positions: {e}")
|
|
|
|
|
|
# ==========================================
|
|
# USAGE EXAMPLE
|
|
# ==========================================
|
|
|
|
"""
|
|
INTEGRATION IN NOTEBOOK:
|
|
|
|
# Cell: Advanced Position Management Setup
|
|
|
|
from advanced_position_management import AdvancedPositionManager, AdaptivePositionSizer
|
|
|
|
# Initialize Manager
|
|
adv_position_mgr = AdvancedPositionManager(
|
|
enable_adaptive_sizing=True,
|
|
enable_trailing_stop=True,
|
|
enable_partial_tp=True
|
|
)
|
|
|
|
print("✅ Advanced Position Management activated!")
|
|
|
|
|
|
# Cell: In execute_trade_v2_adaptive()
|
|
|
|
# BEFORE (old):
|
|
volume = 0.10 # Fixed
|
|
|
|
# AFTER (with Adaptive Sizing):
|
|
if adv_position_mgr.adaptive_sizing:
|
|
volume = adv_position_mgr.adaptive_sizing.calculate_position_size(
|
|
confidence=confidence,
|
|
balance=account_info.balance,
|
|
stop_loss_distance=adjusted_atr_mult * atr * 10000, # Convert to pips
|
|
symbol=symbol
|
|
)
|
|
else:
|
|
volume = 0.10
|
|
|
|
|
|
# Cell: Add to Scheduler (for Trailing Stop + Partial TP)
|
|
|
|
scheduler.add_job(
|
|
func=lambda: adv_position_mgr.check_and_update_positions(symbol),
|
|
trigger='interval',
|
|
minutes=1,
|
|
id='advanced_position_management'
|
|
)
|
|
|
|
print("✅ Advanced Position Management scheduler added!")
|
|
"""
|