""" Live Telegram signal trader. Listens to one or two gold-signal Telegram channels in real time and processes each message with the EXACT same parsing logic as traderanalyser.py: fully detailed signals are taken as given, informal "Gold buy now" messages get the same default SL / TP ladder, and the same precursor-merge logic prevents a "now" announcement and its detailed follow-up from being treated as two trades. Two modes: --dry (DEFAULT - safe): logs what the bot WOULD do for every signal, and tracks a paper position against live market data so you can watch the simulated equity curve evolve. No order is ever sent in this mode. --live : places REAL orders, using the exact same MCP REST calls your live_xgboost.py already uses successfully on both BITMT501 (MT5/Vantage) and bitrtc01 (cTrader/Pepperstone) - same /order/open, /order/modify and /positions endpoints, same request/response shape. Since the channel's signals carry a 4-level TP ladder but a single MT5/cTrader position only has one SL/TP pair, each signal opens one position PER TP level (same lot/n, same SL, each with its own TP) - exactly how scaling-out is normally done. Once the first tranche closes, the remaining tranches' SL is moved to breakeven via /order/modify, same as your _move_sl_to_be. Position sizing defaults to the risk-percentage model (risk_based_lot), not the fixed lot-per-$1000 table - see RISK_PCT_TIERS in trade_signal_analyzer.py for why: the fixed table ignores how wide a given trade's stop is, which is what let one wide-stop signal wipe your backtest account in a single trade. Pass --sizing fixed to match the backtester's original table instead, if you want a direct comparison. Dual-channel support: Pass --channel2-id to also listen on a second Telegram channel that uses the HM XAUUSD format (different message syntax, German BE phrases, REENTRY/RUNNING skipping). The second channel shares the same MT5 server, sizing config and survival rule as channel 1 - the two traders just maintain separate open-position books. Setup: pip install telethon python-dotenv Create a .env file next to this script with: API_ID=... API_HASH=... PHONE_NUMBER=... CHANNEL_ID=-1001234567890 # Goldex (channel 1) CHANNEL2_ID=-1009876543210 # HM XAUUSD (channel 2, optional) Usage: python3 live_signal_trader.py # dry-run, mid tier, risk_pct sizing python3 live_signal_trader.py --risk-tier low python3 live_signal_trader.py --sizing fixed --risk-tier mid python3 live_signal_trader.py --live --mt5-base-url http://localhost:8000 # REAL ORDERS python3 live_signal_trader.py --channel2-id -1009876543210 # + HM channel """ import argparse import json import logging import os import sys from collections import deque from datetime import datetime, timezone, timedelta import requests from dotenv import load_dotenv from telethon import TelegramClient, events from traderanalyser import ( TradeSignalAnalyzer, RISK_TIERS, RISK_PCT_TIERS, CONTRACT_SIZE, DEFAULT_CONTRACT_SIZE, PIP_SIZE, DEFAULT_PIP_SIZE, risk_based_lot, ) try: from hm_signal_parser import HMTradeSignalAnalyzer except ImportError: # nur für --channel2-id nötig HMTradeSignalAnalyzer = None logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') logger = logging.getLogger('live_signal_trader') def _signal_to_dict(signal): """Return a JSON-serialisable copy of a parsed signal dict.""" if signal is None: return None d = {} for k, v in signal.items(): if hasattr(v, 'isoformat'): d[k] = v.isoformat() elif isinstance(v, set): # reported_tp_hits is a set of int indices - render as sorted list d[k] = sorted(v) else: d[k] = v return d load_dotenv() API_ID = os.getenv('API_ID') API_HASH = os.getenv('API_HASH') PHONE_NUMBER = os.getenv('PHONE_NUMBER') CHANNEL_ID_ENV = os.getenv('CHANNEL_ID') CHANNEL2_ID_ENV = os.getenv('CHANNEL2_ID') class LiveSignalTrader: """Wraps a TradeSignalAnalyzer (used purely for its parsing/default- fill logic and MT5 fetch helper - parse_signals()/analyze_trades() from the backtester are never called here) with live position tracking against fresh market data. Survival-sizing rule -------------------- Before opening every trade the bot fetches the REAL account balance from /account and asks: "could I survive 3 consecutive full stop-outs at this trade's stop distance?" survival_lot = balance / (3 * stop_distance * contract_size) If survival_lot < min_lot (0.01 by default): the account is too small to safely run the full TP ladder - the bot opens a SINGLE position of exactly 0.01 lot and exits at TP1 only, so one bad string of three signals won't wipe the account even at minimum size. If survival_lot >= min_lot: normal sizing applies (configured sizing_mode / risk_tier), capped at survival_lot so a single stop-out can never cost more than 1/3 of the current balance. """ def __init__(self, dry_run=True, sizing_mode='risk_pct', risk_tier='mid', initial_capital=100, mt5_base_url='http://localhost:8000', informal_grace_seconds=90, min_lot=0.01, require_sl_tp=False, analyzer_class=None, channel_name='ch1', max_concurrent=1, diagnostics_path=None): self.dry_run = dry_run self.sizing_mode = sizing_mode self.risk_tier = risk_tier self.capital = initial_capital self.informal_grace_seconds = informal_grace_seconds self.min_lot = min_lot self.require_sl_tp = require_sl_tp # when True, skip informal signals entirely self.channel_name = channel_name # used in log prefixes for dual-channel setups self.max_concurrent = max_concurrent # max total open positions across all siblings self._siblings = [] # other LiveSignalTrader instances sharing the same account # --- Diagnostics ring-buffer (last 24 h of messages + trade decisions) --- # Each entry: {'ts': ISO str, 'raw_text': str, 'parsed': signal|None, # 'decision': 'trade'|'skipped(concurrent)'|'informal_pending'| # 'superseded'|'no_signal'|'resolve_failed', 'detail': str} self._diag: deque = deque() # unbounded; pruned to 24 h on each write self.diagnostics_path = diagnostics_path # file to write; None = don't write analyzer_class = analyzer_class or TradeSignalAnalyzer self.analyzer = analyzer_class(file_path=None, initial_capital=initial_capital, mt5_base_url=mt5_base_url) self.open_positions = [] # paper (or real) positions currently being tracked # An informal "Gold buy now" is often just the precursor to a # detailed follow-up seconds later (exactly like your Jan 22 # sample). In file replay we can look at the whole conversation # at once and drop the precursor after the fact; live, we have # to decide immediately and can't undo an order already sent - # so an informal signal is held here for informal_grace_seconds # before actually being opened, and a detailed signal for the # same pair+direction arriving in that window supersedes it. self._pending_informal = {} # (pair, action) -> signal dict, while still within its grace period self._superseded = set() # id() of informal signals that got superseded before their grace period elapsed # ------------------------------------------------------------------ # Diagnostics helpers # ------------------------------------------------------------------ def _diag_record(self, ts, raw_text, parsed_signal, decision, detail=''): """Append one message+decision entry to the ring buffer.""" entry = { 'ts': ts.isoformat() if hasattr(ts, 'isoformat') else str(ts), 'channel': self.channel_name, 'raw_text': raw_text, 'parsed': _signal_to_dict(parsed_signal) if parsed_signal else None, 'decision': decision, 'detail': detail, } self._diag.append(entry) def dump_diagnostics(self, path=None): """Write the last 24 h of messages and trade decisions to a JSON file. Call this any time (e.g. via --dump-diagnostics or from the periodic check loop). Returns the path written.""" out_path = path or self.diagnostics_path if out_path is None: out_path = f'diagnostics_{self.channel_name}.json' cutoff = datetime.now(timezone.utc) - timedelta(hours=24) recent = [e for e in self._diag if datetime.fromisoformat(e['ts'].replace('Z', '+00:00')) >= cutoff] payload = { 'channel': self.channel_name, 'generated_at': datetime.now(timezone.utc).isoformat(), 'window_hours': 24, 'total_messages': len(recent), 'entries': recent, } with open(out_path, 'w', encoding='utf-8') as f: json.dump(payload, f, indent=2, default=str) logger.info(f"[{self.channel_name}] Diagnostics written → {out_path} ({len(recent)} entries)") return out_path def register_sibling(self, other): """Link two traders that share the same MT5 account so the concurrent-position guard counts open trades from both channels.""" if other not in self._siblings: self._siblings.append(other) if self not in other._siblings: other._siblings.append(self) def _total_open(self): """Total open positions across this trader and all siblings. Positions that have already moved to BE (be_done=True) are at zero risk and do NOT block a new concurrent entry.""" at_risk = sum(1 for p in self.open_positions if not p.get('be_done', False)) at_risk += sum( sum(1 for p in s.open_positions if not p.get('be_done', False)) for s in self._siblings ) return at_risk # ------------------------------------------------------------------ # Incoming messages # ------------------------------------------------------------------ def handle_message(self, date, text): """Returns the parsed signal (if any) so the caller can schedule the grace-period finalization for informal ones - see finalize_informal() and the event handler in main().""" logger.info(f"MSG [{date}]: {text!r}") try: signal = self.analyzer.process_message(date, text) except Exception: logger.exception("Failed to parse message - ignoring it") self._diag_record(date, text, None, 'parse_error') return None if signal is None: # Management / update messages (TP done, SL hit, partial close) self._diag_record(date, text, None, 'no_signal', 'management/update message - no new position') return None if signal == 'CANCEL': # Admin retracted the last signal - close any open position for it self._diag_record(date, text, None, 'cancel', 'admin retracted last signal - closing position if open') logger.warning(f"[{self.channel_name}] CANCEL message received: {text!r}") self._handle_cancel(date) return None if signal == 'BE_MOVE': # Trader said "moved all to breakeven" - relay SL → entry to MT5 self._diag_record(date, text, None, 'be_move', 'trader moved SL to breakeven - adjusting open tickets') logger.info(f"[{self.channel_name}] BE_MOVE: {text!r}") self._handle_be_move() return None key = (signal['pair'], signal['action']) if not signal['is_informal']: pending = self._pending_informal.pop(key, None) if pending is not None: self._superseded.add(id(pending)) logger.info(f"Detailed signal for {key} supersedes its informal precursor - " f"only the detailed one will be opened") self._diag_record(date, text, pending, 'superseded', f'informal {key} superseded by this detailed signal') self._open_signal(signal) return signal # Check if we already have an open informal position for this pair+direction. # If so, upgrade it with the real SL/TP rather than skipping as concurrent. informal_pos = next( (p for p in self.open_positions if p['signal'].get('pair') == signal['pair'] and p['signal'].get('action') == signal['action'] and p['signal'].get('is_informal')), None ) if informal_pos is not None: self._upgrade_informal_position(informal_pos, signal, date, text) return signal self._open_signal(signal) else: if self.require_sl_tp: # Strict mode: only trade signals that carry a real SL and TP1. # Informal signals ("Gold sell now") are silently dropped here. self._diag_record(date, text, signal, 'skipped(informal)', 'require_sl_tp=True: informal signal ignored (no SL/TP)') logger.info(f"[{self.channel_name}] Skipping informal {signal['pair']} " f"{signal['action']} - require_sl_tp is active") return None self._pending_informal[key] = signal self._diag_record(date, text, signal, 'informal_pending', f'waiting {self.informal_grace_seconds}s for detailed follow-up') return signal def _handle_be_move(self): """Trader said 'moved all to breakeven'. Move SL to entry on every open ticket for this channel so our MT5 orders mirror the trader's manual adjustment.""" if not self.open_positions: logger.info(f"[{self.channel_name}] BE_MOVE - no open positions to adjust") return for pos in self.open_positions: # Use actual fill price if available; fall back to signal entry midpoint. # signal['entry'] is the ZONE midpoint (e.g. 3903.75 for zone 3902–3905.5), # but the real fill may be well above/below that — using midpoint as BE # puts SL on the wrong side of the fill and guarantees a loss. entry = pos.get('fill_price') or pos['signal'].get('entry') if entry is None: logger.warning(f"[{self.channel_name}] BE_MOVE - no entry price on position, skipping") continue logger.info(f"[{self.channel_name}] BE_MOVE - using BE price {entry:.2f} " f"(fill_price={pos.get('fill_price')}, signal_entry={pos['signal'].get('entry')})") if self.dry_run: logger.info(f"[{self.channel_name}] BE_MOVE [DRY] - would move SL → {entry} " f"for tickets {pos.get('tickets', [])}") pos['current_sl'] = entry else: for ticket in pos.get('tickets', []): self._move_sl_to_be(ticket, entry) pos['current_sl'] = entry def _upgrade_informal_position(self, pos, detailed_signal, date, raw_text): """A detailed signal arrived after we already opened on the informal precursor. Update the open position's SL and TP to match the real signal, and relay the changes to MT5 via /order/modify so the live orders are corrected. The position lot and number of tranches are NOT changed — we can't add new orders after the fact, so we just correct the levels on the existing ticket(s). """ old_sl = pos['signal'].get('sl') old_tps = list(pos['signal'].get('tp_levels', [])) new_sl = detailed_signal.get('sl') new_tps = detailed_signal.get('tp_levels', []) # Build a summary of what changed for logging / diagnostics changes = [] if new_sl and old_sl and abs(new_sl - old_sl) > 0.01: changes.append(f"SL {old_sl:.2f}→{new_sl:.2f}") if new_tps and new_tps != old_tps: changes.append(f"TPs {[round(t,2) for t in old_tps]}→{[round(t,2) for t in new_tps]}") change_str = ' '.join(changes) if changes else 'no price change' logger.info(f"[{self.channel_name}] UPGRADE informal→detailed: {change_str}") # Update the position record pos['signal']['sl'] = new_sl if new_sl else old_sl if new_tps: pos['signal']['tp_levels'] = new_tps if new_sl: pos['current_sl'] = new_sl pos['signal']['is_informal'] = False # mark as upgraded pos['upgraded_from_informal'] = True if self.dry_run: logger.info(f"[{self.channel_name}] UPGRADE [DRY] – position updated in memory; " f"tickets would be modified on live account") else: tickets = pos.get('tickets', []) for i, ticket in enumerate(tickets): # Assign the TP for this tranche (index i into the new TP list) tp = new_tps[i] if new_tps and i < len(new_tps) else None payload = {"ticket": int(ticket)} if new_sl: payload["sl"] = round(float(new_sl), 2) if tp: payload["tp"] = round(float(tp), 2) try: resp = requests.post( f"{self.analyzer.mt5_base_url}/order/modify", json=payload, timeout=10, ) result = resp.json() if result.get('success'): logger.info(f"[LIVE] UPGRADE ticket=#{ticket} {change_str}") else: logger.warning(f"[LIVE] UPGRADE FAILED ticket=#{ticket}: {result.get('error')}") except Exception as e: logger.warning(f"[LIVE] UPGRADE error ticket=#{ticket}: {e}") self._diag_record( date, raw_text, detailed_signal, 'upgraded', f'informal position upgraded: {change_str} tickets={pos.get("tickets", [])} ' f'{"DRY" if self.dry_run else "LIVE"}' ) def _handle_cancel(self, date): """Admin retracted the last signal. Discard any pending informal for the same channel, and close any open position that was opened from the most recent signal (dry: drop it; live: send /order/close).""" # Clear any pending informal (it was the signal being cancelled) self._pending_informal.clear() if not self.open_positions: logger.info(f"[{self.channel_name}] CANCEL - no open positions to close") return # Close the most recently opened position pos = self.open_positions[-1] pair = pos['signal']['pair'] action = pos['signal']['action'] if self.dry_run: self.open_positions.pop() logger.warning( f"[{self.channel_name}] CANCEL [DRY] - dropped position " f"{pair} {action} opened at {pos['signal']['entry']}" ) else: tickets = pos.get('tickets', []) for ticket in tickets: try: resp = requests.post( f"{self.analyzer.mt5_base_url}/order/close", json={"ticket": int(ticket)}, timeout=10, ) result = resp.json() if result.get('success'): logger.warning(f"[{self.channel_name}] CANCEL [LIVE] closed ticket=#{ticket}") else: logger.error( f"[{self.channel_name}] CANCEL [LIVE] close FAILED " f"ticket=#{ticket}: {result.get('error')}" ) except Exception as e: logger.error(f"[{self.channel_name}] CANCEL [LIVE] close error ticket=#{ticket}: {e}") self.open_positions.pop() def finalize_informal(self, signal): """Called by the caller's scheduled task after informal_grace_seconds. If a detailed signal for the same pair+direction superseded this one in the meantime, it's discarded; otherwise it's opened now.""" if id(signal) in self._superseded: self._superseded.discard(id(signal)) # already recorded as 'superseded' in handle_message return key = (signal['pair'], signal['action']) if self._pending_informal.get(key) is signal: del self._pending_informal[key] self._open_signal(signal) def _fetch_account_balance(self): """Fetch the live account balance from the MT5/cTrader MCP server's /account endpoint. Returns the balance as a float, or None on any error (the caller then keeps the last known self.capital value). The endpoint shape matches the MCP server used by live_xgboost.py: GET /account -> {"balance": 1234.56, "equity": 1230.00, ...} """ try: resp = requests.get(f"{self.analyzer.mt5_base_url}/account", timeout=10) resp.raise_for_status() data = resp.json() # Accept both lowercase and title-case keys just in case balance = data.get('balance') or data.get('Balance') if balance is not None and float(balance) > 0: return float(balance) logger.warning(f"/account returned an unexpected payload: {data} - using last known balance") except Exception as e: logger.warning(f"Could not fetch /account: {e} - using last known balance {self.capital:.2f}") return None def _open_signal(self, signal): if not self._resolve_missing_fields(signal): logger.warning(f"Could not resolve a price for {signal['pair']} {signal['action']} " f"at {signal['date']} - skipping, no MT5 data available") self._diag_record(signal['date'], '', signal, 'resolve_failed', 'no live price from MT5 - signal skipped') return # --- Cross-channel concurrent-position guard ------------------------- # When two channels share one account the survival rule's "3 stop-outs" # assumption breaks down if both channels are in a trade at the same # time. Refuse to open a new position once total_open >= max_concurrent. total_open = self._total_open() if self.max_concurrent > 0 and total_open >= self.max_concurrent: sibling_info = ( ' (siblings: ' + ', '.join(f"{s.channel_name}={len(s.open_positions)}" for s in self._siblings) + ')' ) if self._siblings else '' logger.warning( f"[{self.channel_name}] SKIP {signal['pair']} {signal['action']}: " f"total_open={total_open} >= max_concurrent={self.max_concurrent}" f"{sibling_info} — signal ignored to protect account" ) self._diag_record(signal['date'], '', signal, 'skipped(concurrent)', f'total_open={total_open} >= max_concurrent={self.max_concurrent}') return # --- Refresh balance from the live account before every trade --- live_balance = self._fetch_account_balance() if live_balance is not None: if abs(live_balance - self.capital) > 0.01: logger.info(f"Balance updated from API: {self.capital:.2f} -> {live_balance:.2f}") self.capital = live_balance # --- 3-signal survival rule ------------------------------------------ # survival_lot is the largest lot where THREE consecutive full stop-outs # would exhaust at most the entire current balance: # 3 * survival_lot * contract_size * stop_distance = balance # => survival_lot = balance / (3 * stop_distance * contract_size) # which is exactly risk_based_lot(balance, 1/3, ...) # Snapshot the parsed TP list BEFORE survival mode may truncate it, # so the diagnostics always show what the parser actually extracted. _parsed_tp_levels = list(signal['tp_levels']) survival_lot = risk_based_lot(self.capital, 1.0 / 3.0, signal['entry'], signal['sl'], signal['pair']) if survival_lot < self.min_lot: # WARMUP MODE: balance is too small for risk-based sizing of the # full TP ladder, but we calculate how many min-lot tranches the # 3-survival rule can still support at the absolute minimum lot: # # warmup_tranches = floor(survival_lot / min_lot) # = floor(balance / (3 * stop_dist * contract_size * min_lot)) # # This means: # $30 balance → survival_lot ≈ 0.0095 → 0 → capped to 1 TP # $50 balance → survival_lot ≈ 0.016 → 1 TP # $70 balance → survival_lot ≈ 0.022 → 2 TPs # $110 balance → survival_lot ≈ 0.035 → 3 TPs # $140 balance → survival_lot ≈ 0.044 → 4 TPs # (exact thresholds depend on stop distance of each trade) # # The system gradually scales up from TP1 toward the full ladder # as capital grows, instead of staying locked at TP1-only until # suddenly the account is large enough for full risk-based sizing. warmup_tranches = max(1, int(survival_lot / self.min_lot)) n_tranches = warmup_tranches actual_tps = min(warmup_tranches, len(signal['tp_levels'])) # lot = total across all tranches so _place_order splits evenly: # round(actual_tps * min_lot / actual_tps, 2) = min_lot per tranche. # Setting lot = min_lot caused tranche_lot = round(0.01/3,2) = 0.00 # and _place_order returned [] (no orders placed). lot = actual_tps * self.min_lot if actual_tps < len(signal['tp_levels']): logger.warning( f"{signal['pair']} {signal['action']}: balance={self.capital:.2f}, " f"survival lot={survival_lot:.4f} < min={self.min_lot} " f"→ WARMUP MODE: {self.min_lot} lot × {actual_tps} TP(s) " f"(skipping TP{actual_tps+1}..TP{len(signal['tp_levels'])})" ) else: logger.info( f"{signal['pair']} {signal['action']}: balance={self.capital:.2f}, " f"survival lot={survival_lot:.4f} < min={self.min_lot} " f"→ WARMUP MODE: {self.min_lot} lot × {actual_tps} TP(s) (all parsed TPs fit)" ) signal['tp_levels'] = signal['tp_levels'][:actual_tps] else: # Normal sizing, capped at the survival threshold so a single # trade can never cost more than 1/3 of the current balance. configured_lot = self._lot_for(signal) if configured_lot > survival_lot: logger.info( f"{signal['pair']} {signal['action']}: configured lot {configured_lot:.4f} " f"exceeds survival cap {survival_lot:.4f} (1/3 of {self.capital:.2f}) " f"→ capping to survival lot" ) raw_lot = min(configured_lot, survival_lot) if raw_lot < self.min_lot: # Configured risk_pct is so conservative that the resulting lot # is below min_lot even though survival_lot allows more. # Use the same warmup ladder logic: open as many min-lot tranches # as the survival budget supports, so tight-SL signals with a # healthy balance still run the full (or partial) TP ladder # instead of being forced to a single TP by the round-up. # Use survival_lot as total position size (Kelly-optimal): # each tranche = floor(survival_lot / n / min_lot) × min_lot # so we use the full 1/3-of-balance safety budget, not just # n × min_lot. This gives geometric growth as balance rises: # $101→$200 same pace, $200→$300 2× faster, $300→ 3×, etc. warmup_tranches = max(1, int(survival_lot / self.min_lot)) actual_tps = min(warmup_tranches, len(signal['tp_levels'])) n_tranches = actual_tps # tranche_lot = floor to valid lot step (0.01), total ≤ survival_lot tranche_lot_raw = survival_lot / actual_tps tranche_lot = int(tranche_lot_raw / self.min_lot) * self.min_lot tranche_lot = max(tranche_lot, self.min_lot) lot = actual_tps * tranche_lot if actual_tps < len(signal['tp_levels']): logger.warning( f"{signal['pair']} {signal['action']}: configured lot {raw_lot:.4f} < min={self.min_lot}, " f"survival budget allows {warmup_tranches} tranche(s) " f"→ WARMUP(survival): {tranche_lot:.2f} lot × {actual_tps} TP(s) " f"(survival_lot={survival_lot:.4f})" ) else: logger.info( f"{signal['pair']} {signal['action']}: configured lot {raw_lot:.4f} < min={self.min_lot} " f"→ WARMUP(survival): {tranche_lot:.2f} lot × {actual_tps} TP(s) " f"(survival_lot={survival_lot:.4f})" ) signal['tp_levels'] = signal['tp_levels'][:actual_tps] else: lot, n_tranches = self._resolve_tradable_size(raw_lot, len(signal['tp_levels'])) if lot <= 0: logger.warning(f"Computed lot ({raw_lot:.4f}) is below the minimum tradable lot " f"({self.min_lot}) even as a single position - skipping " f"{signal['pair']} {signal['action']}") return if n_tranches < len(signal['tp_levels']): step = len(signal['tp_levels']) / n_tranches signal['tp_levels'] = [signal['tp_levels'][min(int(i * step), len(signal['tp_levels']) - 1)] for i in range(n_tranches)] logger.warning(f"{signal['pair']} {signal['action']}: lot {raw_lot:.4f} too small to split " f"across the full TP ladder at min lot {self.min_lot} - consolidated to " f"{n_tranches} tranche(s): {[round(t, 2) for t in signal['tp_levels']]}") tag = 'DRY' if self.dry_run else 'LIVE' _wt = max(1, int(survival_lot / self.min_lot)) _configured_lot_for_mode = self._lot_for(signal) if survival_lot >= self.min_lot else 0 if survival_lot < self.min_lot or _configured_lot_for_mode < self.min_lot: mode = f'warmup({len(signal["tp_levels"])}tp/0.01)' else: mode = f'{self.sizing_mode}/{self.risk_tier}' logger.info(f"[{tag}] OPEN {signal['action']} {signal['pair']} @ {signal['entry']:.2f} " f"SL {signal['sl']:.2f} TP {[round(t, 2) for t in signal['tp_levels']]} " f"lot={lot:.4f} balance={self.capital:.2f} " f"sizing={mode} ({'informal/defaulted' if signal['is_informal'] else 'detailed'})") import copy as _copy signal_snapshot = _copy.copy(signal) signal_snapshot['tp_levels'] = _parsed_tp_levels # all TPs as parsed, before survival truncation signal_snapshot['tp_levels_used'] = list(signal['tp_levels']) # TPs actually traded self._diag_record(signal['date'], '', signal_snapshot, 'trade', f"lot={lot:.4f} mode={mode} balance={self.capital:.2f} " f"parsed_tps={len(_parsed_tp_levels)} used_tps={len(signal['tp_levels'])} " f"{'DRY' if self.dry_run else 'LIVE'}") if self.dry_run: position = { 'signal': signal, 'lot': lot, 'remaining': list(range(len(signal['tp_levels']))), 'current_sl': signal['sl'], 'last_checked': signal['date'].replace(tzinfo=None), 'realized_pnl': 0.0, } self.open_positions.append(position) else: tickets = self._place_order(signal, lot) if not tickets: logger.error(f"No tranches were actually opened for {signal['pair']} {signal['action']} - " f"not tracking this signal") return # Fetch actual fill prices so BE uses real fill, not signal entry midpoint fill_price = self._fetch_avg_fill(tickets, signal['entry']) logger.info(f"[LIVE] avg fill price for BE: {fill_price:.2f} " f"(signal entry midpoint was {signal['entry']:.2f})") self.open_positions.append({ 'signal': signal, 'tickets': list(tickets), 'n_original': len(tickets), 'be_done': False, 'fill_price': fill_price, }) def _resolve_tradable_size(self, raw_lot, n_desired_tranches): """Enforce the broker's minimum tradable lot (default 0.01). If the raw risk-based lot is below min_lot even as a single position, returns (0, 0) - this signal genuinely can't be traded at this account size and stop width. If it IS tradable as one position but too small to split across the full TP ladder, the ladder gets consolidated down to however many tranches DO fit. Rounding raw_lot UP to min_lot means the actual risk on that trade exceeds your configured risk_pct - an unavoidable consequence of trading a small account against a wide stop, not a bug, and it's logged clearly when it happens.""" if raw_lot <= 0: return 0.0, 0 total_lot = max(raw_lot, self.min_lot) if raw_lot < self.min_lot: logger.warning(f"Lot {raw_lot:.4f} is below the minimum tradable lot ({self.min_lot}) - " f"rounding up, so actual risk here exceeds the configured " f"'{self.risk_tier}' risk_pct") max_tranches = max(1, int(round(total_lot / self.min_lot, 6))) n_tranches = min(n_desired_tranches, max_tranches) return total_lot, n_tranches def _resolve_missing_fields(self, signal): """Same default-fill logic as the backtester's _resolve_informal_signals, just resolved against a live price instead of historical bars.""" if signal['entry'] is None: price = self._get_live_price(signal['pair'], signal['action']) if price is None: return False signal['entry'] = price pip = PIP_SIZE.get(signal['pair'], DEFAULT_PIP_SIZE) if signal['sl'] is None: dist = self.analyzer.default_sl_pips * pip signal['sl'] = signal['entry'] - dist if signal['action'] == 'BUY' else signal['entry'] + dist if not signal['tp_levels']: levels = [signal['entry'] + (d * pip if signal['action'] == 'BUY' else -d * pip) for d in self.analyzer.default_tp_ladder_pips] signal['tp_levels'] = sorted(levels, reverse=(signal['action'] == 'SELL')) return True def _lot_for(self, signal): if self.sizing_mode == 'risk_pct': lot = risk_based_lot(self.capital, RISK_PCT_TIERS[self.risk_tier], signal['entry'], signal['sl'], signal['pair']) else: lot = (self.capital / 1000.0) * RISK_TIERS[self.risk_tier] return round(lot, 4) def _get_live_price(self, pair, action=None): """Live quote via the confirmed /tick/{symbol} endpoint (same one your MT5/cTrader MCP servers expose). Uses ask for BUY, bid for SELL - falls back to the latest M1 bar's close if the tick endpoint is unavailable for some reason.""" try: resp = requests.get(f"{self.analyzer.mt5_base_url}/tick/{pair}", timeout=10) resp.raise_for_status() tick = resp.json() bid, ask = tick.get('bid'), tick.get('ask') if bid and ask: return float(ask) if action == 'BUY' else float(bid) except Exception as e: logger.warning(f"/tick/{pair} failed ({e}) - falling back to latest M1 close") df = self.analyzer._mt5_fetch_ohlc(pair, 'M1', bars=1) if df is None or df.empty: return None return float(df.iloc[-1]['Close']) # ------------------------------------------------------------------ # Position monitoring # ------------------------------------------------------------------ def check_open_positions(self): """Call periodically (see the polling loop in main()). Dry-run: advances the local paper-trading simulation using fresh bars (same fill rules as the backtester's _simulate_trade). Live: polls the real /positions endpoint - a tranche disappearing means it closed (TP or SL), and once the first tranche is gone the remaining ones' SL is moved to breakeven via /order/modify, exactly like your _move_sl_to_be.""" if self.dry_run: self._check_open_positions_dry() else: self._check_open_positions_live() def _check_open_positions_dry(self): still_open = [] for pos in self.open_positions: try: closed = self._advance_position(pos) except Exception: logger.exception(f"Error checking position {pos['signal']['pair']} " f"{pos['signal']['action']} - leaving it open") closed = False if not closed: still_open.append(pos) self.open_positions = still_open def _check_open_positions_live(self): try: resp = requests.get(f"{self.analyzer.mt5_base_url}/positions", timeout=10) resp.raise_for_status() live_tickets = {p['ticket'] for p in resp.json().get('positions', [])} except Exception as e: logger.warning(f"Could not fetch /positions: {e} - leaving positions as tracked") return still_open = [] for pos in self.open_positions: before = len(pos['tickets']) pos['tickets'] = [t for t in pos['tickets'] if t in live_tickets] closed_now = before - len(pos['tickets']) if closed_now: logger.info(f"[LIVE] {closed_now} tranche(s) closed for {pos['signal']['pair']} " f"{pos['signal']['action']} ({len(pos['tickets'])} still open)") if not pos['tickets']: logger.info(f"[LIVE] All tranches closed for {pos['signal']['pair']} {pos['signal']['action']}") continue # fully closed - drop tracking if not pos['be_done'] and len(pos['tickets']) < pos['n_original']: # at least one tranche closed (the first TP, in the normal # case) - move the remaining tranches' SL to breakeven. # Use actual fill price (stored at open), NOT signal entry # midpoint: if fill was 3996.73 but signal entry was 3998.50, # setting BE to 3998.50 puts SL above fill → instant loss. be_price = pos.get('fill_price', pos['signal']['entry']) for ticket in pos['tickets']: self._move_sl_to_be(ticket, be_price) pos['be_done'] = True still_open.append(pos) self.open_positions = still_open def _fetch_avg_fill(self, tickets, fallback_entry): """Query /positions right after placement to get actual fill prices. Returns the average open price across the given tickets, or fallback_entry if the API call fails or no matching positions found.""" try: resp = requests.get(f"{self.analyzer.mt5_base_url}/positions", timeout=10) resp.raise_for_status() positions = resp.json().get('positions', []) ticket_set = set(int(t) for t in tickets) fills = [float(p.get('open_price') or p.get('price_open')) for p in positions if int(p.get('ticket', -1)) in ticket_set and (p.get('open_price') or p.get('price_open'))] if fills: avg = sum(fills) / len(fills) logger.info(f"[LIVE] fill prices from /positions: {fills} → avg {avg:.2f}") return round(avg, 2) except Exception as e: logger.warning(f"[LIVE] _fetch_avg_fill failed ({e}) - using signal entry {fallback_entry:.2f}") return float(fallback_entry) def _move_sl_to_be(self, ticket, entry_price): """Move SL to breakeven via /order/modify - same call and same payload shape as your _move_sl_to_be in live_xgboost.py.""" entry_rounded = round(float(entry_price), 2) try: resp = requests.post(f"{self.analyzer.mt5_base_url}/order/modify", json={ "ticket": int(ticket), "sl": entry_rounded, }, timeout=10) result = resp.json() if result.get('success'): logger.info(f"[LIVE] BE activated ticket=#{ticket} SL -> {entry_rounded}") else: logger.warning(f"[LIVE] BE FAILED ticket=#{ticket}: {result.get('error')}") except Exception as e: logger.warning(f"[LIVE] BE error ticket=#{ticket}: {e}") def _advance_position(self, pos): signal = pos['signal'] df = self.analyzer._mt5_fetch_ohlc(signal['pair'], 'M1', bars=1000) if df is None or df.empty: return False new_bars = df[df.index > pos['last_checked']] if new_bars.empty: return False action = signal['action'] entry_price = signal['entry'] tp_levels = signal['tp_levels'] contract_size = CONTRACT_SIZE.get(signal['pair'], DEFAULT_CONTRACT_SIZE) tranche = (pos['lot'] * contract_size) / max(len(tp_levels), 1) tag = 'DRY' if self.dry_run else 'LIVE' for ts, bar in new_bars.iterrows(): high, low, close = float(bar['High']), float(bar['Low']), float(bar['Close']) sl_triggered = (action == 'BUY' and close <= pos['current_sl']) or \ (action == 'SELL' and close >= pos['current_sl']) if sl_triggered and pos['remaining']: for _ in pos['remaining']: pos['realized_pnl'] += tranche * ((pos['current_sl'] - entry_price) if action == 'BUY' else (entry_price - pos['current_sl'])) label = 'BE' if pos['current_sl'] == entry_price else 'SL' pos['remaining'] = [] self._close_position(pos, label) return True for idx in list(pos['remaining']): tp_price = tp_levels[idx] hit = (action == 'BUY' and high >= tp_price) or (action == 'SELL' and low <= tp_price) if hit: pos['realized_pnl'] += tranche * ((tp_price - entry_price) if action == 'BUY' else (entry_price - tp_price)) pos['remaining'].remove(idx) logger.info(f"[{tag}] TP{idx + 1} hit on {signal['pair']} {action} @ {tp_price:.2f} " f"running pnl={pos['realized_pnl']:.2f}") if pos['current_sl'] != entry_price: pos['current_sl'] = entry_price # breakeven after first confirmed TP pos['last_checked'] = ts if not pos['remaining']: self._close_position(pos, 'ALL_TP') return True return False def _close_position(self, pos, label): tag = 'DRY' if self.dry_run else 'LIVE' self.capital = max(0.0, self.capital + pos['realized_pnl']) logger.info(f"[{tag}] CLOSE {pos['signal']['pair']} {pos['signal']['action']} ({label}) " f"pnl={pos['realized_pnl']:.2f} capital={self.capital:.2f}") # ------------------------------------------------------------------ # Real execution - confirmed against your live_xgboost.py's own calls # ------------------------------------------------------------------ def _place_order(self, signal, lot): """Opens one real position per TP level (same lot/n each, same SL, each with its own TP) via /order/open - matches your _execute_order_mt5: POST {symbol, side, volume, sl, tp, comment}, response {"success": bool, "order": , ...}. Returns the list of tickets that were actually opened (a partial list if one tranche's request failed - logged either way).""" n = len(signal['tp_levels']) tranche_lot = round(lot / n, 2) if tranche_lot <= 0: logger.error(f"Tranche lot rounds to 0.00 (lot={lot}, {n} tranches) - " f"capital too low for this stop distance, skipping real order") return [] side = 'buy' if signal['action'] == 'BUY' else 'sell' tickets = [] for tp in signal['tp_levels']: try: resp = requests.post(f"{self.analyzer.mt5_base_url}/order/open", json={ "symbol": signal['pair'], "side": side, "volume": tranche_lot, "sl": round(signal['sl'], 2), "tp": round(tp, 2), "comment": "telegram-signal", }, timeout=10) result = resp.json() except Exception as e: logger.error(f"[LIVE] order/open request failed for tp={tp}: {e}") continue if result.get('success'): ticket = result.get('order') tickets.append(ticket) logger.info(f"[LIVE] order placed ticket=#{ticket} {side} {tranche_lot} {signal['pair']} " f"sl={signal['sl']:.2f} tp={tp:.2f}") else: logger.error(f"[LIVE] order/open FAILED for tp={tp}: " f"{result.get('error')} - {result.get('description')}") return tickets def main(): parser = argparse.ArgumentParser(description="Live Telegram signal trader") parser.add_argument('--dry', action='store_true', help='Explicit dry-run flag (this is also the default if neither flag is given)') parser.add_argument('--live', action='store_true', help='Disable dry-run and place REAL orders via /order/open on --mt5-base-url') parser.add_argument('--sizing', choices=['risk_pct', 'fixed'], default='risk_pct', help="Position sizing model (default risk_pct - see module docstring for why)") parser.add_argument('--risk-tier', choices=['high', 'mid', 'low'], default='mid') parser.add_argument('--initial-capital', type=float, default=100) parser.add_argument('--mt5-base-url', default='http://localhost:8000') parser.add_argument('--channel-id', type=int, default=int(CHANNEL_ID_ENV) if CHANNEL_ID_ENV else None, help='Telegram channel ID for channel 1 (Goldex format). ' 'Defaults to CHANNEL_ID env var.') parser.add_argument('--channel2-id', type=int, default=int(CHANNEL2_ID_ENV) if CHANNEL2_ID_ENV else None, help='Optional second channel (HM XAUUSD format). ' 'Defaults to CHANNEL2_ID env var. ' 'When set, a separate trader instance with the HM parser listens here.') parser.add_argument('--poll-interval', type=int, default=10, help='Seconds between checks of open positions against fresh market data') parser.add_argument('--informal-grace-seconds', type=int, default=90, help='How long to hold an informal ("buy now") signal before opening it, ' 'in case a detailed follow-up supersedes it. ' 'Default 90s: ch1-Goldex detailed signals typically arrive 30-60s after ' 'the informal "buy/sell now" precursor (observed in diagnostics).') parser.add_argument('--min-lot', type=float, default=0.01, help="Broker's minimum tradable lot - the TP ladder is consolidated to fewer " "tranches (or the signal skipped) if the risk-based lot can't meet this") parser.add_argument('--channels', choices=['ch1', 'ch2', 'both'], default='both', help="Which channel(s) to listen on. " "'ch1' = Goldex only, 'ch2' = HM only (requires --channel2-id), " "'both' = both channels (default). " "When channel2-id is not set 'both' and 'ch2' fall back to ch1.") parser.add_argument('--max-concurrent', type=int, default=1, help="Maximum total open positions allowed across ALL active channels at once. " "A new signal is skipped (with a warning) when this limit is already reached. " "Default 1 = never open a second position while one is already open. " "Set to 0 to disable the guard entirely.") parser.add_argument('--diagnostics-dir', default='.', help="Directory where diagnostics_.json files are written. " "Default: current directory. Files are updated every --diag-interval seconds " "and also on clean shutdown (Ctrl-C).") parser.add_argument('--diag-interval', type=int, default=300, help="Seconds between automatic diagnostics file writes (default 300 = every 5 min). " "Set to 0 to disable periodic writes (file is still written on shutdown).") parser.add_argument('--require-sl-tp', action='store_true', default=False, help="Skip informal signals ('Gold sell now') entirely - only open trades when " "a full signal with SL and at least TP1 is present. " "Eliminates auto-generated SL/TP guesses but may miss fast-moving entries.") args = parser.parse_args() if args.channel_id is None and args.channels in ('ch1', 'both') and not args.channel2_id: logger.error("No channel ID given - pass --channel-id or set CHANNEL_ID in .env") sys.exit(1) if not (API_ID and API_HASH and PHONE_NUMBER): logger.error("API_ID / API_HASH / PHONE_NUMBER must be set in .env (same as your existing Telethon script)") sys.exit(1) dry_run = not args.live if not dry_run: print(f"LIVE mode will place REAL orders on {args.mt5_base_url} - one position per TP level, " f"real money at risk.") import sys as _sys if _sys.stdin.isatty(): confirm = input("Type 'yes' to continue: ") if confirm.strip().lower() != 'yes': print("Aborted.") return else: print("Non-interactive mode — live trading auto-confirmed.") logger.info("LIVE mode auto-confirmed (non-interactive / nohup)") # Resolve which channels to actually activate run_ch1 = args.channels in ('ch1', 'both') and args.channel_id is not None run_ch2 = args.channels in ('ch2', 'both') and bool(args.channel2_id) if run_ch2 and HMTradeSignalAnalyzer is None: logger.error("--channel2-id gesetzt, aber hm_signal_parser.py fehlt im Image") sys.exit(1) if args.channels in ('ch2', 'both') and not args.channel2_id: logger.warning("--channels includes ch2 but no --channel2-id / CHANNEL2_ID set — ch2 disabled") if not run_ch1 and not run_ch2: logger.error("No active channel after applying --channels filter — nothing to listen to") sys.exit(1) os.makedirs(args.diagnostics_dir, exist_ok=True) def _diag_path(channel_name): return os.path.join(args.diagnostics_dir, f'diagnostics_{channel_name}.json') # --- Channel 1: Goldex (TradeSignalAnalyzer) -------------------------------- trader = None if run_ch1: trader = LiveSignalTrader(dry_run=dry_run, sizing_mode=args.sizing, risk_tier=args.risk_tier, initial_capital=args.initial_capital, mt5_base_url=args.mt5_base_url, informal_grace_seconds=args.informal_grace_seconds, min_lot=args.min_lot, require_sl_tp=args.require_sl_tp, channel_name='ch1-Goldex', max_concurrent=args.max_concurrent, diagnostics_path=_diag_path('ch1-Goldex')) # --- Channel 2: HM XAUUSD (HMTradeSignalAnalyzer) ------------------------- trader2 = None if run_ch2: trader2 = LiveSignalTrader(dry_run=dry_run, sizing_mode=args.sizing, risk_tier=args.risk_tier, initial_capital=args.initial_capital, mt5_base_url=args.mt5_base_url, informal_grace_seconds=args.informal_grace_seconds, min_lot=args.min_lot, require_sl_tp=args.require_sl_tp, analyzer_class=HMTradeSignalAnalyzer, channel_name='ch2-HM', max_concurrent=args.max_concurrent, diagnostics_path=_diag_path('ch2-HM')) # Link siblings so each trader's concurrent guard sees the other's positions if trader is not None and trader2 is not None: trader.register_sibling(trader2) # --- Startup balance fetch -------------------------------------------------- # One account → fetch once and apply to whichever traders are active. ref_trader = trader or trader2 # at least one is not None (checked above) startup_balance = ref_trader._fetch_account_balance() for t in (trader, trader2): if t is not None: if startup_balance is not None: t.capital = startup_balance # else: keeps args.initial_capital as seed if startup_balance is not None: logger.info(f"Account balance from API: {startup_balance:.2f} USDT " f"({'seed' if startup_balance == args.initial_capital else 'live'} value)") else: logger.warning(f"Could not reach {args.mt5_base_url}/account - " f"using seed capital {args.initial_capital:.2f} USDT for sizing until the " f"first signal fires (will retry before each trade)") client = TelegramClient(os.getenv('TG_SESSION', 'mysession'), API_ID, API_HASH) # --- Event handlers -------------------------------------------------------- if trader is not None: @client.on(events.NewMessage(chats=args.channel_id)) async def handler_ch1(event): import asyncio try: date = event.message.date # tz-aware UTC datetime from Telethon text = event.message.text or '' signal = trader.handle_message(date, text) if signal is not None and signal['is_informal']: async def delayed_finalize(): await asyncio.sleep(args.informal_grace_seconds) trader.finalize_informal(signal) client.loop.create_task(delayed_finalize()) except Exception: logger.exception("[ch1-Goldex] Error processing incoming message") if trader2 is not None: @client.on(events.NewMessage(chats=args.channel2_id)) async def handler_ch2(event): import asyncio try: date = event.message.date text = event.message.text or '' signal = trader2.handle_message(date, text) if signal is not None and signal['is_informal']: async def delayed_finalize2(): await asyncio.sleep(args.informal_grace_seconds) trader2.finalize_informal(signal) client.loop.create_task(delayed_finalize2()) except Exception: logger.exception("[ch2-HM] Error processing incoming message") def dump_all_diagnostics(): """Write diagnostics files for all active traders.""" for t in (trader, trader2): if t is not None: try: t.dump_diagnostics() except Exception as e: logger.warning(f"Could not write diagnostics for {t.channel_name}: {e}") async def periodic_check(): import asyncio last_diag = 0.0 while True: await asyncio.sleep(args.poll_interval) # Refresh balance once; both traders share the same account. live_bal = ref_trader._fetch_account_balance() if live_bal is not None: for t in (trader, trader2): if t is not None: t.capital = live_bal tag = 'DRY' if dry_run else 'LIVE' parts = [] if trader is not None: parts.append(f"ch1_open={len(trader.open_positions)}") if trader2 is not None: parts.append(f"ch2_open={len(trader2.open_positions)}") logger.info(f"[{tag}] Balance: {live_bal:.2f} USDT " + " ".join(parts) + f" total_open={ref_trader._total_open()}" f" max_concurrent={args.max_concurrent}") for t in (trader, trader2): if t is not None: t.check_open_positions() # Periodic diagnostics dump import time now = time.monotonic() if args.diag_interval > 0 and (now - last_diag) >= args.diag_interval: dump_all_diagnostics() last_diag = now # ------------------------------------------------------------------ # Start Telethon and enter the live-listener loop # ------------------------------------------------------------------ client.start(phone=PHONE_NUMBER) client.loop.create_task(periodic_check()) active = [] if trader is not None: active.append(f"ch1={args.channel_id} [Goldex]") if trader2 is not None: active.append(f"ch2={args.channel2_id} [HM XAUUSD]") logger.info( f"Active channels: {', '.join(active)} | " f"{'DRY-RUN' if dry_run else 'LIVE'} " f"(sizing={args.sizing}, risk_tier={args.risk_tier}, " f"seed_capital={args.initial_capital}, min_lot={args.min_lot}, " f"max_concurrent={args.max_concurrent}) " f"Survival rule: balance fetched live before each trade; " f"if survival_lot < {args.min_lot} → warmup mode (0.01 lot, TPs scaled to balance)" ) logger.info(f"Diagnostics: writing to {args.diagnostics_dir}/ every " f"{args.diag_interval}s (also on Ctrl-C shutdown)") try: client.run_until_disconnected() finally: # Always write a final diagnostics snapshot on shutdown so the # last session is available for analysis even if no periodic write # fired yet. dump_all_diagnostics() logger.info("Shutdown complete - diagnostics saved.") if __name__ == '__main__': main()