""" position_manager.py Manages already-open trades so a winner doesn't have to give back all of its gains just to hit the original stop loss, and reconciles the trade log once MT5 shows a position has actually closed (whether via SL, TP, or a manual close from the dashboard) — without this reconciliation, nothing ever marked a trade "closed" in the database, so graduation stats could never populate. Runs independently of the candle-close cycle (checked every Config.POSITION_MANAGE_INTERVAL_SECONDS, default 0.1s) since a trade can move meaningfully in far less time than an H4/D1 candle takes to close. Rules, all risk-REDUCING by construction — the stop only ever moves in the trade's favor, never back toward more risk: 1. Breakeven move at BREAKEVEN_TRIGGER_R, once profit clears a DYNAMIC buffer that's the largest of: a flat R-fraction, the symbol's current ATR-scaled distance, and the round-trip trading cost itself (spread + commission) — so a real move can't be erased by ordinary noise or costs eating into "breakeven." 2. Trailing stop from TRAILING_ACTIVATION_R, locking in at least (profit_R - TRAILING_DISTANCE_R). The effective trailing distance tightens (via TRAILING_TIGHTEN_FACTOR) if a HIGHER timeframe's market structure has flipped against the trade's direction — top-down context narrowing an already-safe action, never loosening it. 3. Time decay: past TIME_DECAY_HOURS without reaching breakeven, the stop gradually tightens toward entry (capped at TIME_DECAY_MAX_TIGHTEN_PCT of the current SL-to-entry distance) — a trade that hasn't proven its setup gets less benefit of the doubt the longer it sits open. Only ever touches positions carrying WickBot's own magic number. """ import logging import threading import time from datetime import datetime, timezone from config import Config import performance_tracker as perf import indicators from wick_rules import market_structure_bias log = logging.getLogger("wickbot.position_manager") EPSILON = 1e-6 # float-comparison tolerance around R thresholds TIMEFRAME_UP_MAP = { "M1": "M15", "M5": "M30", "M15": "H1", "M30": "H4", "H1": "H4", "H4": "D1", "D1": "D1", } _warned_missing_tickets = set() _volatility_cache = {} # symbol -> (fetched_at, atr_value) _bias_cache = {} # symbol -> (fetched_at, bias_str) _stats_lock = threading.Lock() _stats = { "managed_positions": 0, "breakeven_moves": 0, "trail_adjustments": 0, "time_decay_tightens": 0, "closed_by_manager": 0, } def _increment_stat(key: str, amount: int = 1): with _stats_lock: _stats[key] += amount def reset_management_stats(): with _stats_lock: for k in _stats: _stats[k] = 0 def get_management_stats() -> dict: with _stats_lock: snapshot = dict(_stats) snapshot["magic_number"] = Config.MAGIC_NUMBER return snapshot def log_management_summary(): stats = get_management_stats() log.info( "Position manager summary | managed=%d breakeven_moves=%d trail_adjustments=%d " "time_decay_tightens=%d closed_by_manager=%d magic=%d", stats["managed_positions"], stats["breakeven_moves"], stats["trail_adjustments"], stats["time_decay_tightens"], stats["closed_by_manager"], stats["magic_number"], ) # --------------------------------------------------------------------- # Core math # --------------------------------------------------------------------- def _current_price_for_direction(tick, side: str) -> float: # Closing a buy happens at bid; closing a sell happens at ask. return tick.bid if side == "buy" else tick.ask def _profit_r(entry: float, initial_sl: float, current_price: float, side: str) -> float: risk_distance = abs(entry - initial_sl) if risk_distance <= 0: return 0.0 if side == "buy": return (current_price - entry) / risk_distance return (entry - current_price) / risk_distance def _get_atr(connector, symbol: str) -> float: """Small, cached ATR fetch — avoids hammering the terminal for a fresh candle pull every 30-second cycle for every open position.""" now = time.time() cached = _volatility_cache.get(symbol) if cached and (now - cached[0]) < Config.VOLATILITY_CACHE_SECONDS: return cached[1] try: df = connector.get_candles(symbol, Config.TIMEFRAME, n=30) atr_series = indicators.atr(df, 14) atr_value = float(atr_series.iloc[-1]) except Exception: atr_value = 0.0 _volatility_cache[symbol] = (now, atr_value) return atr_value def _get_higher_tf_bias(connector, symbol: str) -> str: """Cached top-down structure bias on the next timeframe up from Config.TIMEFRAME — 'bullish', 'bearish', or 'range'.""" if not Config.HIGHER_TF_BIAS_ENABLED: return "range" now = time.time() cached = _bias_cache.get(symbol) if cached and (now - cached[0]) < Config.VOLATILITY_CACHE_SECONDS: return cached[1] try: higher_tf = TIMEFRAME_UP_MAP.get(Config.TIMEFRAME.upper(), Config.TIMEFRAME) df = connector.get_candles(symbol, higher_tf, n=100) bias = market_structure_bias(df) except Exception: bias = "range" _bias_cache[symbol] = (now, bias) return bias def _cost_buffer_price(connector, symbol: str, volume: float) -> float: """Converts round-trip trading cost (spread + commission) into a price-distance equivalent, so 'breakeven' means genuinely flat after costs, not just flat on raw price.""" try: sym_info = connector.symbol_info(symbol) except Exception: return 0.0 point = getattr(sym_info, "point", 0.0) or 0.0 spread_points = getattr(sym_info, "spread", 0) or 0 spread_price = (spread_points * point) if Config.INCLUDE_SPREAD_IN_BREAKEVEN else 0.0 commission_price = 0.0 if Config.COMMISSION_PER_LOT_ROUNDTRIP > 0: tick_value = getattr(sym_info, "trade_tick_value", 0.0) or 0.0 if tick_value > 0 and point > 0: # tick_value is $ per point for 1.0 lot; commission is $ per # lot round-trip — volume cancels out of this ratio entirely # (both the cost and the tick value scale with volume the # same way), so this is correct regardless of position size. commission_points = Config.COMMISSION_PER_LOT_ROUNDTRIP / tick_value commission_price = commission_points * point return spread_price + commission_price def _dynamic_buffer_price(connector, symbol: str, volume: float, risk_distance: float) -> float: """The buffer used for the breakeven move — the LARGEST of a flat R-fraction, ATR-scaled volatility, and real trading cost, so the 'breakeven' stop can't be undone by noise, spread, or commission.""" r_based = risk_distance * Config.BREAKEVEN_BUFFER_R atr_based = _get_atr(connector, symbol) * Config.ATR_BUFFER_MULTIPLIER cost_based = _cost_buffer_price(connector, symbol, volume) return max(r_based, atr_based, cost_based) def _hours_since(iso_timestamp: str) -> float: try: opened = datetime.fromisoformat(iso_timestamp) if opened.tzinfo is None: opened = opened.replace(tzinfo=timezone.utc) return (datetime.now(timezone.utc) - opened).total_seconds() / 3600 except Exception: return 0.0 # --------------------------------------------------------------------- # Per-position management # --------------------------------------------------------------------- def _manage_one(connector, position, trade_row: dict): symbol = position.symbol side = trade_row["side"] entry = trade_row["entry"] initial_sl = trade_row["initial_sl"] current_sl = position.sl volume = trade_row.get("volume") or getattr(position, "volume", 0.0) tick = connector.get_tick(symbol) if not tick: return None current_price = _current_price_for_direction(tick, side) profit_r = _profit_r(entry, initial_sl, current_price, side) risk_distance = abs(entry - initial_sl) if risk_distance <= 0: return None new_sl = None breakeven_now = trade_row["breakeven_moved"] trailing_now = trade_row["trailing_active"] time_decay_applied = False # --- Rule 1: breakeven move, with a dynamic (noise/cost-aware) buffer --- if not trade_row["breakeven_moved"] and profit_r >= Config.BREAKEVEN_TRIGGER_R - EPSILON: buffer_price = _dynamic_buffer_price(connector, symbol, volume, risk_distance) candidate = entry + buffer_price if side == "buy" else entry - buffer_price if (side == "buy" and candidate > current_sl) or (side == "sell" and candidate < current_sl): new_sl = candidate breakeven_now = True # --- Rule 2: trailing stop, tightened if the higher-TF bias has flipped --- if profit_r >= Config.TRAILING_ACTIVATION_R - EPSILON: bias = _get_higher_tf_bias(connector, symbol) bias_supports_trade = (bias == "bullish" and side == "buy") or (bias == "bearish" and side == "sell") bias_opposes_trade = (bias == "bullish" and side == "sell") or (bias == "bearish" and side == "buy") effective_distance_r = Config.TRAILING_DISTANCE_R if bias_opposes_trade: effective_distance_r = Config.TRAILING_DISTANCE_R * Config.TRAILING_TIGHTEN_FACTOR locked_r = profit_r - effective_distance_r candidate = entry + locked_r * risk_distance if side == "buy" else entry - locked_r * risk_distance base_sl = new_sl if new_sl is not None else current_sl if (side == "buy" and candidate > base_sl) or (side == "sell" and candidate < base_sl): new_sl = candidate trailing_now = True # --- Rule 3: time decay — tighten a stagnant trade that hasn't reached breakeven --- if new_sl is None and not trade_row["breakeven_moved"]: hours_open = _hours_since(trade_row["opened_at"]) if trade_row.get("opened_at") else 0.0 if hours_open > Config.TIME_DECAY_HOURS: overage = min(1.0, (hours_open - Config.TIME_DECAY_HOURS) / max(Config.TIME_DECAY_HOURS, 1.0)) tighten_pct = overage * Config.TIME_DECAY_MAX_TIGHTEN_PCT gap = entry - current_sl if side == "buy" else current_sl - entry candidate = current_sl + gap * tighten_pct if side == "buy" else current_sl - gap * tighten_pct if (side == "buy" and candidate > current_sl) or (side == "sell" and candidate < current_sl): new_sl = candidate time_decay_applied = True if new_sl is None: return None result = connector.modify_position_sl(position.ticket, symbol, new_sl, tp=trade_row.get("tp")) success = getattr(result, "retcode", None) == 10009 # TRADE_RETCODE_DONE if success: perf.update_trade_sl(trade_row["id"], new_sl, breakeven_moved=breakeven_now, trailing_active=trailing_now) _increment_stat("managed_positions") if breakeven_now and not trade_row["breakeven_moved"]: _increment_stat("breakeven_moves") if trailing_now and not trade_row["trailing_active"]: _increment_stat("trail_adjustments") if time_decay_applied: _increment_stat("time_decay_tightens") log.info( "%s %s: SL -> %.5f (profit=%.2fR, breakeven=%s, trailing=%s, time_decay=%s)", symbol, side, new_sl, profit_r, breakeven_now, trailing_now, time_decay_applied, ) return { "symbol": symbol, "side": side, "new_sl": new_sl, "profit_r": profit_r, "breakeven_moved": breakeven_now and not trade_row["breakeven_moved"], "trailing_active": trailing_now and not trade_row["trailing_active"], "time_decay_applied": time_decay_applied, } log.warning("SL modify rejected for %s ticket=%s", symbol, position.ticket) return None # --------------------------------------------------------------------- # Closed-trade reconciliation # --------------------------------------------------------------------- def _classify_exit_reason(deals, tp, sl, side) -> str: """Compares the actual exit price (the last deal's price) to the trade's stored TP/SL to classify how it closed. Uses a small tolerance since a real fill can land a few points past the exact level (slippage) rather than landing on it precisely.""" if not deals or tp is None or sl is None: return "other" exit_price = getattr(deals[-1], "price", None) if exit_price is None: return "other" tp_distance = abs(tp - sl) * 0.15 # tolerance: 15% of the SL-to-TP span if abs(exit_price - tp) <= tp_distance: return "tp" if abs(exit_price - sl) <= tp_distance: return "sl" return "other" # manual close, trailing stop hit somewhere else, etc. def reconcile_closed_trades(connector): """Finds trades the DB still thinks are 'open' that MT5 no longer shows as open positions (closed via SL, TP, or a manual close), pulls the realized P&L from history, and marks them closed with a result_r computed against the $ actually risked at open. Without this, every graduation/expectancy stat stays stuck at zero forever. CRITICAL FIX: MT5's deal.profit is ALREADY the net realized P&L including commission and swap. Do NOT add commission and swap separately — that would double-count them and could flip the sign of a losing trade to show as positive (e.g. a -$10 net loss with a +$15 commission rebate would incorrectly show as +$5 profit). """ open_tickets = {p.ticket for p in connector.get_bot_positions()} db_open_trades = perf.get_all_open_trades() for row in db_open_trades: if row["ticket"] in open_tickets: continue # still genuinely open try: deals = connector.get_history_deals_for_position(row["ticket"]) except Exception: log.exception("Failed to fetch history deals for ticket=%s", row["ticket"]) continue # MT5 deal.profit is ALREADY net of commission and swap. # Adding commission and swap separately would double-count them, # potentially flipping a loss into a false positive. # Only sum deal.profit — it's the true net realized P&L. total_profit = sum( getattr(d, "profit", 0.0) for d in deals ) risk_amount = row.get("risk_amount") or 0.0 result_r = (total_profit / risk_amount) if risk_amount > 0 else 0.0 status = "closed_win" if total_profit > 0 else ("closed_loss" if total_profit < 0 else "closed_be") exit_reason = _classify_exit_reason(deals, row.get("tp"), row.get("initial_sl"), row["side"]) perf.log_trade_close(row["id"], round(result_r, 3), status, exit_reason=exit_reason) _increment_stat("closed_by_manager") log.info( "Reconciled closed trade: %s ticket=%s profit=%.2f result=%.2fR status=%s exit=%s", row["symbol"], row["ticket"], total_profit, result_r, status, exit_reason, ) # --------------------------------------------------------------------- # Manual controls (used by the dashboard's Positions card) # --------------------------------------------------------------------- def manual_modify(connector, ticket: int, new_sl: float = None, new_tp: float = None): """Manual SL/TP override from the dashboard. Unlike the automatic rules above, this does NOT enforce 'never loosen' — it's an explicit owner action, so it's trusted at face value. Updates the DB record to match afterward.""" trade_row = perf.get_open_trade_by_ticket(ticket) if not trade_row: raise ValueError(f"No open trade record found for ticket {ticket}") sl_to_send = new_sl if new_sl is not None else trade_row["current_sl"] tp_to_send = new_tp if new_tp is not None else trade_row.get("tp") result = connector.modify_position_sl(ticket, trade_row["symbol"], sl_to_send, tp=tp_to_send) success = getattr(result, "retcode", None) == 10009 if success and new_sl is not None: perf.update_trade_sl(trade_row["id"], new_sl) return success def manual_close(connector, ticket: int): """Manual full close from the dashboard. Reconciliation on the next cycle picks up the realized result — called once immediately here too, so the dashboard reflects it right away rather than waiting up to POSITION_MANAGE_INTERVAL_SECONDS.""" trade_row = perf.get_open_trade_by_ticket(ticket) if not trade_row: raise ValueError(f"No open trade record found for ticket {ticket}") position = next((p for p in connector.get_bot_positions() if p.ticket == ticket), None) if not position: raise ValueError(f"No open MT5 position found for ticket {ticket}") result = connector.close_position(ticket, trade_row["symbol"], position.volume, trade_row["side"]) success = getattr(result, "retcode", None) == 10009 if success: reconcile_closed_trades(connector) return success # --------------------------------------------------------------------- # Read-only status + background loop # --------------------------------------------------------------------- def get_status(connector) -> list: """Read-only snapshot of every managed position — current profit in R, breakeven/trailing state, and how long it's been open. Safe to call as often as you like (e.g. /positions in Telegram, or the dashboard's Positions card).""" out = [] for position in connector.get_bot_positions(): trade_row = perf.get_open_trade_by_ticket(position.ticket) if not trade_row: continue tick = connector.get_tick(position.symbol) if not tick: continue current_price = _current_price_for_direction(tick, trade_row["side"]) profit_r = _profit_r(trade_row["entry"], trade_row["initial_sl"], current_price, trade_row["side"]) out.append({ "symbol": position.symbol, "side": trade_row["side"], "ticket": position.ticket, "volume": getattr(position, "volume", trade_row.get("volume")), "entry": trade_row["entry"], "profit_r": round(profit_r, 2), "current_sl": position.sl, "tp": position.tp, "hours_open": round(_hours_since(trade_row["opened_at"]), 1) if trade_row.get("opened_at") else None, "breakeven_moved": trade_row["breakeven_moved"], "trailing_active": trade_row["trailing_active"], }) return out def get_pending_status(connector) -> list: """Read-only snapshot of pending orders (bot only places market orders today, but this covers manually-placed or future ones).""" out = [] for order in connector.get_pending_orders(): out.append({ "ticket": order.ticket, "symbol": order.symbol, "type": order.type, "volume": order.volume_current, "price_open": order.price_open, "sl": order.sl, "tp": order.tp, }) return out def run_once(connector, tg=None): """Single pass: reconcile anything that's closed, then manage anything still open. Returns the list of SL changes made this cycle (for notification purposes).""" try: reconcile_closed_trades(connector) except Exception: log.exception("Closed-trade reconciliation failed") if not Config.POSITION_MANAGEMENT_ENABLED: return [] changes = [] for position in connector.get_bot_positions(): trade_row = perf.get_open_trade_by_ticket(position.ticket) if not trade_row: if position.ticket not in _warned_missing_tickets: log.info( "No matching open trade record for ticket=%s (%s) — skipping management.", position.ticket, position.symbol, ) _warned_missing_tickets.add(position.ticket) continue _increment_stat("managed_positions") try: change = _manage_one(connector, position, trade_row) if change: changes.append(change) except Exception: log.exception("Error managing position ticket=%s", position.ticket) if tg and changes: for c in changes: if c["breakeven_moved"]: tg.notify(f"🔒 {c['symbol']} {c['side'].upper()}: SL moved to breakeven (+{c['profit_r']:.2f}R).") elif c["trailing_active"]: tg.notify(f"📈 {c['symbol']} {c['side'].upper()}: trailing stop advanced (+{c['profit_r']:.2f}R).") elif c.get("time_decay_applied"): tg.notify(f"⏳ {c['symbol']} {c['side'].upper()}: time-decay SL tightened (+{c['profit_r']:.2f}R).") return changes def start_position_manager(connector, tg=None): """Runs in a background thread, independent of the candle-close cycle, so breakeven/trailing moves and closed-trade reconciliation happen promptly rather than waiting on a possibly-hours-away H4/D1 bar close. Runs at Config.POSITION_MANAGE_INTERVAL_SECONDS (default 0.1s) for near-real-time position monitoring. """ def _loop(): while True: try: run_once(connector, tg) except Exception: log.exception("Position manager cycle failed") time.sleep(Config.POSITION_MANAGE_INTERVAL_SECONDS) thread = threading.Thread(target=_loop, daemon=True) thread.start() return thread