From 2c4899bb983f40d212722fb20fa08f07ebc51b69 Mon Sep 17 00:00:00 2001 From: bast Date: Wed, 4 Nov 2026 22:35:18 +0100 Subject: [PATCH] implement --- bots.env | 8 + bots/Dockerfile | 10 + bots/live_signal_trader.py | 1187 ++++++++++++++++++++++++++++++++++++ bots/mcp_server.py | 1000 ++++++++++++++++++++++++++++++ bots/requirements.txt | 12 + bots/traderanalyser.py | 1040 +++++++++++++++++++++++++++++++ docker-compose.yaml | 61 +- 7 files changed, 3313 insertions(+), 5 deletions(-) create mode 100644 bots.env create mode 100644 bots/Dockerfile create mode 100644 bots/live_signal_trader.py create mode 100644 bots/mcp_server.py create mode 100644 bots/requirements.txt create mode 100644 bots/traderanalyser.py diff --git a/bots.env b/bots.env new file mode 100644 index 00000000..10b58442 --- /dev/null +++ b/bots.env @@ -0,0 +1,8 @@ +MT5_LOGIN=… +MT5_PASSWORD=… +MT5_SERVER=… + +API_ID=1234 +API_HASH=abccccccccccccccccccd134 +PHONE_NUMBER=+41796600666 +CHANNEL2_ID=-1234 diff --git a/bots/Dockerfile b/bots/Dockerfile new file mode 100644 index 00000000..5c61667c --- /dev/null +++ b/bots/Dockerfile @@ -0,0 +1,10 @@ +# Ein gemeinsames Image für mcp_server und live_signal_trader (~500 MB statt zwei) +FROM python:3.11-slim + +ENV PYTHONUNBUFFERED=1 PIP_NO_CACHE_DIR=1 +WORKDIR /app + +COPY requirements.txt . +RUN pip install -r requirements.txt + +COPY *.py ./ diff --git a/bots/live_signal_trader.py b/bots/live_signal_trader.py new file mode 100644 index 00000000..3d912ec1 --- /dev/null +++ b/bots/live_signal_trader.py @@ -0,0 +1,1187 @@ +""" +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() diff --git a/bots/mcp_server.py b/bots/mcp_server.py new file mode 100644 index 00000000..aefeb5a7 --- /dev/null +++ b/bots/mcp_server.py @@ -0,0 +1,1000 @@ +#!/usr/bin/env python3 +""" +MT5 MCP Server for Linux/Devuan +================================ +FastAPI-based HTTP server that provides MCP-compatible endpoints +for MetaTrader 5 integration on Linux via Wine + mt5linux. + +Modes: + --mock : Run with simulated data (for testing) + --live : Connect to real MT5 via mt5linux rpyc bridge + +Usage: + python mcp_server.py --mock # Test mode + python mcp_server.py --live # Real MT5 connection + python mcp_server.py --mock --port 8000 +""" + +import os +import sys +import json +import logging +import argparse +import random +import asyncio +from datetime import datetime, timedelta, timezone +from typing import Dict, Any, Optional, List +from contextlib import asynccontextmanager + +from fastapi import FastAPI, HTTPException, Query +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse +from pydantic import BaseModel +import uvicorn + +# --------------------------------------------------------------------------- +# Logging +# --------------------------------------------------------------------------- +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + handlers=[ + logging.StreamHandler(sys.stdout), + logging.FileHandler("/tmp/mt5_mcp_server.log", mode="a"), + ], +) +log = logging.getLogger("mt5-mcp") + +# --------------------------------------------------------------------------- +# MT5 Bridge (abstraction over mock / live) +# --------------------------------------------------------------------------- +class MT5Bridge: + """Base class – subclassed by MockBridge and LiveBridge.""" + + def __init__(self): + self.connected = False + self.mode = "base" + + async def connect(self) -> bool: + raise NotImplementedError + + async def disconnect(self): + self.connected = False + + async def account_info(self) -> Dict[str, Any]: + raise NotImplementedError + + async def symbols(self) -> List[str]: + raise NotImplementedError + + async def symbol_info(self, symbol: str) -> Dict[str, Any]: + raise NotImplementedError + + async def ohlc(self, symbol: str, timeframe: str, bars: int) -> List[Dict]: + raise NotImplementedError + + async def tick(self, symbol: str) -> Dict[str, Any]: + raise NotImplementedError + + async def positions(self) -> List[Dict]: + raise NotImplementedError + + async def orders(self) -> List[Dict]: + raise NotImplementedError + + +# ---- Mock Bridge ---------------------------------------------------------- +class MockBridge(MT5Bridge): + """Generates realistic-looking fake data for testing.""" + + SYMBOLS = [ + "EURUSD", "GBPUSD", "USDJPY", "USDCHF", "AUDUSD", + "USDCAD", "NZDUSD", "XAUUSD", "XAGUSD", "BTCUSD", + ] + BASE_PRICES = { + "EURUSD": 1.0850, "GBPUSD": 1.2650, "USDJPY": 149.50, + "USDCHF": 0.8820, "AUDUSD": 0.6550, "USDCAD": 1.3580, + "NZDUSD": 0.6120, "XAUUSD": 2650.00, "XAGUSD": 31.50, + "BTCUSD": 67500.00, + } + SPREADS = { + "EURUSD": 0.00012, "GBPUSD": 0.00015, "USDJPY": 0.015, + "USDCHF": 0.00014, "AUDUSD": 0.00013, "USDCAD": 0.00016, + "NZDUSD": 0.00015, "XAUUSD": 0.30, "XAGUSD": 0.03, + "BTCUSD": 25.0, + } + + def __init__(self): + super().__init__() + self.mode = "mock" + + async def connect(self) -> bool: + self.connected = True + log.info("MockBridge connected (simulated data)") + return True + + async def account_info(self) -> Dict[str, Any]: + return { + "login": 12345678, + "server": "MockBroker-Demo", + "balance": 10000.00, + "equity": 10025.50, + "margin": 125.00, + "free_margin": 9900.50, + "margin_level": 8020.40, + "leverage": 500, + "currency": "USD", + "name": "Demo Account", + "company": "MockBroker Ltd", + "trade_mode": "DEMO", + } + + async def symbols(self) -> List[str]: + return self.SYMBOLS + + async def symbol_info(self, symbol: str) -> Dict[str, Any]: + if symbol not in self.BASE_PRICES: + raise HTTPException(404, f"Symbol {symbol} not found") + price = self.BASE_PRICES[symbol] + spread = self.SPREADS.get(symbol, 0.0002) + digits = 5 if price < 10 else (3 if price < 200 else (2 if price < 5000 else 1)) + return { + "symbol": symbol, + "bid": round(price - spread / 2, digits), + "ask": round(price + spread / 2, digits), + "spread": spread, + "digits": digits, + "point": 10 ** -digits, + "trade_contract_size": 100000 if "USD" in symbol and "XA" not in symbol else (100 if "XAU" in symbol else 1), + "volume_min": 0.01, + "volume_max": 100.0, + "volume_step": 0.01, + "description": f"{symbol} mock data", + } + + async def ohlc(self, symbol: str, timeframe: str, bars: int, start_pos: int = 0) -> List[Dict]: + """Generate mock OHLC bars.""" + if symbol not in self.BASE_PRICES: + raise HTTPException(404, f"Symbol {symbol} not found") + + tf_seconds = { + "S5": 5, "S10": 10, "S15": 15, "S30": 30, + "M1": 60, "M5": 300, "M15": 900, "M30": 1800, + "H1": 3600, "H4": 14400, "D1": 86400, "W1": 604800, + } + seconds = tf_seconds.get(timeframe, 3600) + price = self.BASE_PRICES[symbol] + volatility = price * (0.0002 if seconds < 60 else 0.001) + + now = datetime.now(tz=None).replace(microsecond=0) + # Align to timeframe + now = now - timedelta(seconds=int(now.timestamp()) % max(seconds, 1)) + + data = [] + current_price = price + for i in range(bars - 1, -1, -1): + bar_time = now - timedelta(seconds=seconds * i) + o = current_price + h = o + random.uniform(0, volatility) + l = o - random.uniform(0, volatility) + c = random.uniform(l, h) + v = random.randint(50, 5000) + digits = 5 if price < 10 else (3 if price < 200 else 2) + data.append({ + "time": bar_time.isoformat() + "Z", + "timestamp": int(bar_time.timestamp()), + "open": round(o, digits), + "high": round(h, digits), + "low": round(l, digits), + "close": round(c, digits), + "volume": v, + }) + current_price = c + return data + + async def tick(self, symbol: str) -> Dict[str, Any]: + if symbol not in self.BASE_PRICES: + raise HTTPException(404, f"Symbol {symbol} not found") + price = self.BASE_PRICES[symbol] + spread = self.SPREADS.get(symbol, 0.0002) + jitter = price * random.uniform(-0.0005, 0.0005) + digits = 5 if price < 10 else (3 if price < 200 else 2) + bid = round(price + jitter - spread / 2, digits) + ask = round(price + jitter + spread / 2, digits) + return { + "symbol": symbol, + "bid": bid, + "ask": ask, + "last": round((bid + ask) / 2, digits), + "volume": random.randint(1, 100), + "time": datetime.now(tz=None).isoformat() + "Z", + } + + async def positions(self) -> List[Dict]: + return [ + { + "ticket": 100001, + "symbol": "EURUSD", + "type": "BUY", + "volume": 0.10, + "open_price": 1.0840, + "current_price": 1.0852, + "profit": 12.00, + "swap": -0.50, + "sl": 1.0800, + "tp": 1.0900, + "open_time": (datetime.now(tz=None) - timedelta(hours=3)).isoformat() + "Z", + "magic": 0, + "comment": "mock position", + } + ] + + async def orders(self) -> List[Dict]: + return [] + + +# ---- Live Bridge (mt5linux) ----------------------------------------------- +class LiveBridge(MT5Bridge): + """Connects to real MT5 via mt5linux rpyc bridge.""" + + def __init__(self, host: str = "localhost", port: int = 8001): + super().__init__() + self.mode = "live" + self.rpyc_host = host + self.rpyc_port = port + self.mt5 = None + self._rpyc_conn = None # persistent rpyc connection for tick data + + def _get_rpyc(self): + """Get or create persistent rpyc connection for direct Wine-Python access.""" + import rpyc + if self._rpyc_conn is None or self._rpyc_conn.closed: + try: + self._rpyc_conn = rpyc.classic.connect(self.rpyc_host, self.rpyc_port) + self._rpyc_conn.execute("import MetaTrader5 as mt5") + self._rpyc_conn.execute("from datetime import datetime, timedelta") + self._rpyc_conn.execute("mt5.initialize()") + log.info("Direct rpyc connection established for tick data") + except Exception as e: + log.error(f"rpyc connection failed: {e}") + self._rpyc_conn = None + raise + return self._rpyc_conn + + async def connect(self) -> bool: + try: + from mt5linux import MetaTrader5 as MT5 + self.mt5 = MT5(host=self.rpyc_host, port=self.rpyc_port) + # Broker-Login aus Umgebungsvariablen (bots.env) – leer = Terminal-Login verwenden + kwargs = {} + if os.getenv("MT5_LOGIN"): + kwargs = { + "login": int(os.getenv("MT5_LOGIN")), + "password": os.getenv("MT5_PASSWORD", ""), + "server": os.getenv("MT5_SERVER", ""), + } + log.info(f"Logging in to MT5 account {kwargs['login']} on {kwargs['server']}") + if not self.mt5.initialize(**kwargs): + error = self.mt5.last_error() + log.error(f"MT5 initialize failed: {error}") + return False + info = self.mt5.account_info() + log.info(f"LiveBridge connected: login={info.login}, server={info.server}") + self.connected = True + return True + except ImportError: + log.error("mt5linux not installed. Run: pip install mt5linux") + return False + except Exception as e: + log.error(f"LiveBridge connection failed: {e}") + return False + + async def disconnect(self): + if self._rpyc_conn: + try: + self._rpyc_conn.close() + except Exception: + pass + self._rpyc_conn = None + if self.mt5: + try: + self.mt5.shutdown() + except Exception: + pass + self.connected = False + + async def account_info(self) -> Dict[str, Any]: + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + info = self.mt5.account_info() + if info is None: + raise HTTPException(500, f"account_info failed: {self.mt5.last_error()}") + return { + "login": info.login, + "server": info.server, + "balance": info.balance, + "equity": info.equity, + "margin": info.margin, + "free_margin": info.margin_free, + "margin_level": info.margin_level, + "leverage": info.leverage, + "currency": info.currency, + "name": info.name, + "company": info.company, + "trade_mode": "DEMO" if info.trade_mode == 0 else "REAL", + } + + async def symbols(self) -> List[str]: + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + syms = self.mt5.symbols_get() + if syms is None: + return [] + return [s.name for s in syms if s.visible] + + async def symbol_info(self, symbol: str) -> Dict[str, Any]: + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + info = self.mt5.symbol_info(symbol) + if info is None: + raise HTTPException(404, f"Symbol {symbol} not found") + return { + "symbol": info.name, + "bid": info.bid, + "ask": info.ask, + "spread": info.spread, + "digits": info.digits, + "point": info.point, + "trade_contract_size": info.trade_contract_size, + "volume_min": info.volume_min, + "volume_max": info.volume_max, + "volume_step": info.volume_step, + "description": info.description, + } + + async def ohlc(self, symbol: str, timeframe: str, bars: int, start_pos: int = 0) -> List[Dict]: + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + import pandas as pd + + # Check if seconds timeframe + seconds_map = {"S5": 5, "S10": 10, "S15": 15, "S30": 30} + if timeframe in seconds_map: + return await self._ohlc_from_ticks(symbol, seconds_map[timeframe], bars) + + tf_map = { + "M1": 1, "M5": 5, "M15": 15, "M30": 30, + "H1": 16385, "H4": 16388, "D1": 16408, "W1": 32769, + } + tf = tf_map.get(timeframe, 16385) # default H1 + rates = self.mt5.copy_rates_from_pos(symbol, tf, start_pos, bars) + if rates is None or len(rates) == 0: + raise HTTPException(404, f"No data for {symbol} {timeframe}") + + df = pd.DataFrame(rates) + df["time"] = pd.to_datetime(df["time"], unit="s") + result = [] + for _, row in df.iterrows(): + result.append({ + "time": row["time"].isoformat() + "Z", + "timestamp": int(row["time"].timestamp()), + "open": float(row["open"]), + "high": float(row["high"]), + "low": float(row["low"]), + "close": float(row["close"]), + "volume": int(row["tick_volume"]), + }) + return result + + + async def ohlc_range(self, symbol: str, timeframe: str, + from_ts: int, to_ts: int) -> List[Dict]: + """ + Hole OHLCV Bars für einen Zeitraum via copy_rates_range. + Ermöglicht chunked Downloads für beliebig lange History. + + Args: + symbol: z.B. "XAUUSD" + timeframe: M1, M5, M15, etc. + from_ts: Unix timestamp Start + to_ts: Unix timestamp Ende + """ + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + import pandas as pd + from datetime import datetime, timezone + + tf_map = { + "M1": 1, "M5": 5, "M15": 15, "M30": 30, + "H1": 16385, "H4": 16388, "D1": 16408, "W1": 32769, + } + tf = tf_map.get(timeframe, 1) + + # Convert timestamps to datetime for MT5 + date_from = datetime.fromtimestamp(from_ts, tz=timezone.utc) + date_to = datetime.fromtimestamp(to_ts, tz=timezone.utc) + + try: + rates = self.mt5.copy_rates_range(symbol, tf, date_from, date_to) + except Exception as e: + # rpyc datetime workaround: pass as naive datetime + try: + date_from_naive = datetime.utcfromtimestamp(from_ts) + date_to_naive = datetime.utcfromtimestamp(to_ts) + rates = self.mt5.copy_rates_range(symbol, tf, date_from_naive, date_to_naive) + except Exception as e2: + raise HTTPException(500, f"copy_rates_range failed: {e2}") + + if rates is None or len(rates) == 0: + return [] + + df = pd.DataFrame(rates) + df["time"] = pd.to_datetime(df["time"], unit="s") + + result = [] + for _, row in df.iterrows(): + result.append({ + "time": row["time"].isoformat() + "Z", + "timestamp": int(row["time"].timestamp()), + "open": float(row["open"]), + "high": float(row["high"]), + "low": float(row["low"]), + "close": float(row["close"]), + "volume": int(row["tick_volume"]), + }) + return result + + async def _ohlc_from_ticks(self, symbol: str, period_seconds: int, bars: int) -> List[Dict]: + """Build sub-minute OHLC bars from real tick data via persistent rpyc.""" + import pandas as pd + + conn = self._get_rpyc() + + # Fetch ticks — all in Wine-Python, only pull converted tuples + ticks_needed = max(bars * 20, 5000) + minutes_back = max(int(bars * period_seconds / 60) * 2, 10) + + conn.execute( + f'_ticks = mt5.copy_ticks_from("{symbol}", ' + f'datetime.utcnow() - timedelta(minutes={minutes_back}), ' + f'{ticks_needed}, mt5.COPY_TICKS_ALL)' + ) + count = conn.eval("len(_ticks)") + log.info(f"Fetched {count} ticks for {symbol} ({minutes_back}min back)") + + if count == 0: + raise HTTPException(404, f"No tick data for {symbol} — market may be closed") + + # Convert on Wine side, pull as plain list + conn.execute( + "_td = [(int(_ticks[i][0]), float(_ticks[i][1]), float(_ticks[i][2])) " + "for i in range(len(_ticks))]" + ) + tick_data = list(conn.eval("_td")) + + # Build DataFrame locally + df = pd.DataFrame(tick_data, columns=["time", "bid", "ask"]) + df["dt"] = pd.to_datetime(df["time"], unit="s") + df["mid"] = (df["bid"] + df["ask"]) / 2 + + # Resample into bars + rule = f"{period_seconds}s" + grouped = df.set_index("dt")["mid"].resample(rule) + ohlc_open = grouped.first().dropna() + ohlc_high = grouped.max().dropna() + ohlc_low = grouped.min().dropna() + ohlc_close = grouped.last().dropna() + vol = grouped.count() + + idx = ohlc_open.index.intersection(ohlc_close.index) + idx = idx.intersection(ohlc_high.index).intersection(ohlc_low.index) + idx = idx[-bars:] + + if len(idx) == 0: + raise HTTPException(404, f"No complete {period_seconds}s bars for {symbol}") + + # Get current tick to fix last (incomplete) bar + try: + conn.execute(f'_curtick = mt5.symbol_info_tick("{symbol}")') + cur_bid = conn.eval("float(_curtick.bid)") + cur_ask = conn.eval("float(_curtick.ask)") + cur_mid = (cur_bid + cur_ask) / 2 + except Exception: + cur_mid = None + + result = [] + last_idx = idx[-1] + for ts in idx: + o = float(ohlc_open[ts]) + h = float(ohlc_high[ts]) + l = float(ohlc_low[ts]) + c = float(ohlc_close[ts]) + + # Fix last bar with current price + if ts == last_idx and cur_mid is not None: + c = cur_mid + if cur_mid > h: + h = cur_mid + if cur_mid < l: + l = cur_mid + + result.append({ + "time": ts.isoformat() + "Z", + "timestamp": int(ts.timestamp()), + "open": round(o, 6), + "high": round(h, 6), + "low": round(l, 6), + "close": round(c, 6), + "volume": int(vol.get(ts, 0)), + }) + log.info(f"Built {len(result)} x {period_seconds}s bars from {len(df)} ticks") + return result + + + async def order_open(self, symbol: str, side: str, volume: float, + sl: float = 0, tp: float = 0, + comment: str = "") -> Dict[str, Any]: + """Open a market order.""" + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + + # Get symbol info for filling + info = self.mt5.symbol_info(symbol) + if info is None: + raise HTTPException(404, f"Symbol {symbol} not found") + + price = info.ask if side == "buy" else info.bid + order_type = 0 if side == "buy" else 1 # ORDER_TYPE_BUY / SELL + + request = { + "action": 1, # TRADE_ACTION_DEAL + "symbol": symbol, + "volume": volume, + "type": order_type, + "price": price, + "sl": sl, + "tp": tp, + "deviation": 20, + "magic": 202604, + "comment": comment or "xgboost_sweep", + "type_time": 0, # ORDER_TIME_GTC + "type_filling": 1, # ORDER_FILLING_IOC + } + + result = self.mt5.order_send(request) + if result is None: + error = self.mt5.last_error() + raise HTTPException(500, f"order_send failed: {error}") + + return { + "retcode": result.retcode, + "deal": result.deal, + "order": result.order, + "volume": result.volume, + "price": result.price, + "comment": result.comment, + "success": result.retcode == 10009, # TRADE_RETCODE_DONE + "error": result.comment if result.retcode != 10009 else None, + } + + async def order_modify(self, ticket: int, sl: float = None, + tp: float = None) -> Dict[str, Any]: + """Modify SL/TP of an existing position.""" + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + + # Get current position + positions = self.mt5.positions_get(ticket=ticket) + if positions is None or len(positions) == 0: + raise HTTPException(404, f"Position {ticket} not found") + + pos = positions[0] + request = { + "action": 6, # TRADE_ACTION_SLTP + "position": ticket, + "symbol": pos.symbol, + "sl": sl if sl is not None else pos.sl, + "tp": tp if tp is not None else pos.tp, + "type_filling": 1, # ORDER_FILLING_IOC (Vantage) + } + + result = self.mt5.order_send(request) + if result is None: + error = self.mt5.last_error() + raise HTTPException(500, f"order_modify failed: {error}") + + return { + "retcode": result.retcode, + "success": result.retcode == 10009, + "error": result.comment if result.retcode != 10009 else None, + } + + async def order_close(self, ticket: int) -> Dict[str, Any]: + """Close an existing position.""" + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + + positions = self.mt5.positions_get(ticket=ticket) + if positions is None or len(positions) == 0: + raise HTTPException(404, f"Position {ticket} not found") + + pos = positions[0] + close_type = 1 if pos.type == 0 else 0 # Reverse: BUY->SELL, SELL->BUY + price = self.mt5.symbol_info_tick(pos.symbol) + close_price = price.bid if pos.type == 0 else price.ask + + request = { + "action": 1, # TRADE_ACTION_DEAL + "symbol": pos.symbol, + "volume": pos.volume, + "type": close_type, + "position": ticket, + "price": close_price, + "deviation": 20, + "magic": 202604, + "comment": "xgboost_close", + "type_time": 0, + "type_filling": 1, + } + + result = self.mt5.order_send(request) + if result is None: + error = self.mt5.last_error() + raise HTTPException(500, f"order_close failed: {error}") + + return { + "retcode": result.retcode, + "deal": result.deal, + "success": result.retcode == 10009, + "error": result.comment if result.retcode != 10009 else None, + } + + + async def tick(self, symbol: str) -> Dict[str, Any]: + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + tick = self.mt5.symbol_info_tick(symbol) + if tick is None: + raise HTTPException(404, f"No tick for {symbol}") + return { + "symbol": symbol, + "bid": tick.bid, + "ask": tick.ask, + "last": tick.last, + "volume": tick.volume, + "time": datetime.fromtimestamp(tick.time, tz=timezone.utc).isoformat() + "Z", + } + + async def positions(self) -> List[Dict]: + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + pos = self.mt5.positions_get() + if pos is None: + return [] + result = [] + for p in pos: + result.append({ + "ticket": p.ticket, + "symbol": p.symbol, + "type": "BUY" if p.type == 0 else "SELL", + "volume": p.volume, + "open_price": p.price_open, + "current_price": p.price_current, + "profit": p.profit, + "swap": p.swap, + "sl": p.sl, + "tp": p.tp, + "open_time": datetime.fromtimestamp(p.time, tz=timezone.utc).isoformat() + "Z", + "magic": p.magic, + "comment": p.comment, + }) + return result + + async def orders(self) -> List[Dict]: + if not self.mt5: + raise HTTPException(503, "MT5 not connected") + ords = self.mt5.orders_get() + if ords is None: + return [] + return [ + { + "ticket": o.ticket, + "symbol": o.symbol, + "type": o.type, + "volume": o.volume_current, + "price": o.price_open, + "sl": o.sl, + "tp": o.tp, + "time_setup": datetime.fromtimestamp(o.time_setup, tz=timezone.utc).isoformat() + "Z", + "comment": o.comment, + } + for o in ords + ] + + +# --------------------------------------------------------------------------- +# Global bridge instance (set in lifespan) +# --------------------------------------------------------------------------- +bridge: MT5Bridge = MockBridge() + + +@asynccontextmanager +async def lifespan(app: FastAPI): + global bridge + ok = await bridge.connect() + # Im Container startet MT5 beim ersten Mal mehrere Minuten – so lange warten statt + # dauerhaft "degraded" zu laufen. Healthcheck bleibt rot bis die Verbindung steht. + while not ok and bridge.mode == "live": + log.warning("MT5 not ready yet – retrying in 15s") + await asyncio.sleep(15) + ok = await bridge.connect() + if ok: + log.info(f"Server starting in {bridge.mode.upper()} mode") + else: + log.warning(f"Bridge connect failed ({bridge.mode}) – server runs but data may be unavailable") + yield + await bridge.disconnect() + log.info("Server shutdown complete") + + +# --------------------------------------------------------------------------- +# FastAPI App +# --------------------------------------------------------------------------- +app = FastAPI( + title="MT5 MCP Server", + version="3.0.0", + description="Linux MCP server for MetaTrader 5 – mock or live mode", + lifespan=lifespan, +) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +start_time = datetime.now(tz=None) + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + +@app.get("/") +async def root(): + return { + "service": "MT5 MCP Server", + "version": "3.0.0", + "mode": bridge.mode, + "connected": bridge.connected, + "uptime_s": int((datetime.now(tz=None) - start_time).total_seconds()), + "endpoints": [ + "GET /health", + "GET /account", + "GET /symbols", + "GET /symbol/{symbol}", + "GET /ohlc/{symbol}?timeframe=H1&bars=100", + "GET /tick/{symbol}", + "GET /positions", + "GET /orders", + "GET /mcp/tools", + ], + } + + +@app.get("/health") +async def health(): + return { + "status": "ok" if bridge.connected else "degraded", + "mode": bridge.mode, + "connected": bridge.connected, + "uptime_s": int((datetime.now(tz=None) - start_time).total_seconds()), + "timestamp": datetime.now(tz=None).isoformat() + "Z", + } + + +@app.get("/account") +async def account(): + return await bridge.account_info() + + +@app.get("/symbols") +async def symbols(): + syms = await bridge.symbols() + return {"count": len(syms), "symbols": syms} + + +@app.get("/symbol/{symbol}") +async def symbol_info(symbol: str): + return await bridge.symbol_info(symbol.upper()) + + +@app.get("/ohlc/{symbol}") +async def ohlc( + symbol: str, + timeframe: str = Query("H1", pattern="^(S5|S10|S15|S30|M1|M5|M15|M30|H1|H4|D1|W1)$"), + bars: int = Query(100, ge=1, le=10000), + start: int = Query(0, ge=0, description="Offset: 0=latest, 10000=next page back"), +): + data = await bridge.ohlc(symbol.upper(), timeframe, bars, start_pos=start) + return { + "symbol": symbol.upper(), + "timeframe": timeframe, + "bars": len(data), + "data": data, + } + + + +@app.get("/ohlc_range/{symbol}") +async def ohlc_range( + symbol: str, + timeframe: str = Query("M1", pattern="^(M1|M5|M15|M30|H1|H4|D1|W1)$"), + from_ts: int = Query(..., description="Unix timestamp start"), + to_ts: int = Query(..., description="Unix timestamp end"), +): + """Fetch OHLCV bars for a specific time range (chunked downloads).""" + data = await bridge.ohlc_range(symbol.upper(), timeframe, from_ts, to_ts) + return { + "symbol": symbol.upper(), + "timeframe": timeframe, + "from": from_ts, + "to": to_ts, + "bars": len(data), + "data": data, + } + + + +# --------------------------------------------------------------------------- +# Order Execution Endpoints +# --------------------------------------------------------------------------- + +class OrderRequest(BaseModel): + symbol: str + side: str # "buy" or "sell" + volume: float = 0.01 + sl: float = 0 + tp: float = 0 + comment: str = "" + +class ModifyRequest(BaseModel): + ticket: int + sl: float = None + tp: float = None + +class CloseRequest(BaseModel): + ticket: int + + +@app.post("/order/open") +async def order_open(req: OrderRequest): + """Open a market order.""" + if req.side not in ("buy", "sell"): + raise HTTPException(400, "side must be 'buy' or 'sell'") + return await bridge.order_open( + req.symbol.upper(), req.side, req.volume, + req.sl, req.tp, req.comment + ) + + +@app.post("/order/modify") +async def order_modify(req: ModifyRequest): + """Modify SL/TP of existing position.""" + return await bridge.order_modify(req.ticket, req.sl, req.tp) + + +@app.post("/order/close") +async def order_close(req: CloseRequest): + """Close existing position.""" + return await bridge.order_close(req.ticket) + + +@app.get("/tick/{symbol}") +async def tick(symbol: str): + return await bridge.tick(symbol.upper()) + + +@app.get("/positions") +async def positions(): + pos = await bridge.positions() + return {"count": len(pos), "positions": pos} + + +@app.get("/orders") +async def orders(): + ords = await bridge.orders() + return {"count": len(ords), "orders": ords} + + +# --------------------------------------------------------------------------- +# MCP Tool Definitions (for AI integration / n8n) +# --------------------------------------------------------------------------- +@app.get("/mcp/tools") +async def mcp_tools(): + """Return MCP-compatible tool definitions.""" + return { + "tools": [ + { + "name": "get_account_info", + "description": "Get MT5 trading account info (balance, equity, margin, leverage)", + "inputSchema": {"type": "object", "properties": {}}, + }, + { + "name": "get_symbols", + "description": "List available trading symbols", + "inputSchema": {"type": "object", "properties": {}}, + }, + { + "name": "get_symbol_info", + "description": "Get details for a specific symbol (bid, ask, spread, contract size)", + "inputSchema": { + "type": "object", + "properties": {"symbol": {"type": "string", "description": "Symbol name, e.g. EURUSD"}}, + "required": ["symbol"], + }, + }, + { + "name": "get_ohlc", + "description": "Get OHLC candlestick data for a symbol", + "inputSchema": { + "type": "object", + "properties": { + "symbol": {"type": "string", "description": "Symbol name"}, + "timeframe": {"type": "string", "enum": ["M1", "M5", "M15", "M30", "H1", "H4", "D1", "W1"]}, + "bars": {"type": "integer", "description": "Number of bars (1-10000)", "default": 100}, + }, + "required": ["symbol"], + }, + }, + { + "name": "get_tick", + "description": "Get current tick (bid/ask) for a symbol", + "inputSchema": { + "type": "object", + "properties": {"symbol": {"type": "string"}}, + "required": ["symbol"], + }, + }, + { + "name": "get_positions", + "description": "Get open trading positions", + "inputSchema": {"type": "object", "properties": {}}, + }, + { + "name": "get_orders", + "description": "Get pending orders", + "inputSchema": {"type": "object", "properties": {}}, + }, + ] + } + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- +def main(): + global bridge + + parser = argparse.ArgumentParser(description="MT5 MCP Server") + parser.add_argument("--mock", action="store_true", help="Use mock data (default)") + parser.add_argument("--live", action="store_true", help="Connect to real MT5 via mt5linux") + parser.add_argument("--host", default="0.0.0.0", help="Listen host (default: 0.0.0.0)") + parser.add_argument("--port", type=int, default=8000, help="Listen port (default: 8000)") + parser.add_argument("--rpyc-host", default="localhost", help="mt5linux rpyc host") + parser.add_argument("--rpyc-port", type=int, default=8001, help="mt5linux rpyc port") + args = parser.parse_args() + + if args.live: + bridge = LiveBridge(host=args.rpyc_host, port=args.rpyc_port) + else: + bridge = MockBridge() + + log.info(f"Starting MT5 MCP Server on {args.host}:{args.port} [{bridge.mode} mode]") + uvicorn.run(app, host=args.host, port=args.port, log_level="info") + + +if __name__ == "__main__": + main() diff --git a/bots/requirements.txt b/bots/requirements.txt new file mode 100644 index 00000000..aaea470c --- /dev/null +++ b/bots/requirements.txt @@ -0,0 +1,12 @@ +# MCP-Server +fastapi +uvicorn +pydantic +mt5linux +rpyc +pandas +numpy +# Live-Trader +requests +telethon +python-dotenv diff --git a/bots/traderanalyser.py b/bots/traderanalyser.py new file mode 100644 index 00000000..3fed4f90 --- /dev/null +++ b/bots/traderanalyser.py @@ -0,0 +1,1040 @@ +import pandas as pd +import re +import statistics +from datetime import datetime +import requests + + +# Map however the provider writes a pair/asset name to a canonical code. +PAIR_ALIASES = { + 'GOLD': 'XAUUSD', 'XAU': 'XAUUSD', 'XAUUSD': 'XAUUSD', + 'SILVER': 'XAGUSD', 'XAG': 'XAGUSD', 'XAGUSD': 'XAGUSD', + 'BTC': 'BTCUSD', 'BITCOIN': 'BTCUSD', 'BTCUSD': 'BTCUSD', +} + +# Canonical pair -> symbol name as known by your MT5 terminal/broker. +# Identity mapping for almost everything (your broker uses "XAUUSD" etc. +# directly - confirmed via your /ohlc/XAUUSD sweep). Only fill this in if +# a particular broker uses a suffixed symbol (e.g. "XAUUSD.a"). +MT5_SYMBOL_OVERRIDES = { + # 'XAUUSD': 'XAUUSD.a', +} + +# Timeframes to try, finest first, falling back to coarser ones until one +# actually reaches back far enough to cover a given signal's date. Based +# on your own /ohlc history-depth sweep on BITMT501 (M1 ~7 days, M5 ~2 +# months, M15 ~5 months, H1 ~2 years, H4 ~7 years, D1 ~back to 2007). +MT5_TIMEFRAME_FALLBACK_CHAIN = ['M15', 'H1', 'H4', 'D1'] + +# Lot size per $1000 of account equity, taken directly from your own +# money-management table ($100->0.10/0.05/0.01, $200->0.20/0.10/0.02, +# $500->0.50/0.25/0.05, $1000->1.00/0.50/0.10 - all the same ratio). +# +# IMPORTANT CAVEAT: this sizes lot purely off capital and ignores how far +# away the stop is for any given trade. Most of your signals use a tight +# stop, but the sanity check allows entries up to PRICE_SANITY_PCT (12%) +# from entry, and a handful of real signals do have much wider stops than +# the 60-pip default. At the high tier, lot = capital/1000, so a stop +# that's $480 away (12% of a $4000 gold price) loses 480*lot*100 = 48x +# capital in that ONE trade - which is exactly how the high-tier run hit +# exactly $0.00. This table is kept for comparison; RISK_PCT_TIERS below +# is the safer alternative actually used by the live trader. +RISK_TIERS = { + 'high': 1.00, + 'mid': 0.50, + 'low': 0.10, +} + +# Safer alternative: risk a FIXED percentage of current capital on every +# trade, with lot size computed from that trade's ACTUAL stop distance +# rather than a one-size-fits-all table. This caps the worst case loss on +# any single trade at risk_pct of capital, no matter how wide that +# particular trade's stop happens to be - eliminating the single-trade +# wipeout risk that RISK_TIERS carries. +RISK_PCT_TIERS = { + 'high': 0.03, # risk 3% of capital per trade + 'mid': 0.015, # 1.5% + 'low': 0.005, # 0.5% +} + + +def risk_based_lot(capital, risk_pct, entry, sl, pair): + """lot size such that a full stop-out loses exactly risk_pct of + capital, regardless of how wide this particular trade's stop is.""" + contract_size = CONTRACT_SIZE.get(pair, DEFAULT_CONTRACT_SIZE) + stop_distance = abs(entry - sl) + if stop_distance <= 0: + return 0.0 + return (capital * risk_pct) / (stop_distance * contract_size) + +# Contract size (units per 1.00 lot) used to convert a price move into a +# dollar P&L: pnl = lot * contract_size * price_difference. 100 oz/lot is +# the common CFD convention for XAUUSD on brokers like Vantage/Pepperstone +# - double-check yours via the MT5 symbol specification if unsure. +CONTRACT_SIZE = { + 'XAUUSD': 100, + 'XAGUSD': 5000, + 'BTCUSD': 1, +} +DEFAULT_CONTRACT_SIZE = 100_000 # standard FX lot, units of base currency + +# Price distance of one "pip" per pair, used to translate your stated +# default risk plan (60 pip SL, 20/40/60/80 pip TP ladder) into actual +# price levels for informal signals ("Gold buy now") that don't carry +# their own SL/TP. Derived from your real signals' TP spacing ($2 per 20 +# pips on XAUUSD, i.e. pip=$0.1). +PIP_SIZE = { + 'XAUUSD': 0.1, + 'XAGUSD': 0.01, + 'BTCUSD': 1.0, +} +DEFAULT_PIP_SIZE = 0.0001 # standard FX pairs + +# Smallest tradable lot at most retail brokers (Vantage/Pepperstone both +# use 0.01). Used by the 'survival' sizing mode to decide when to fall +# back to the minimal-size / TP1-only path. +MIN_LOT = 0.01 + +# If an informal "now" signal is followed within this many seconds by a +# fully detailed signal for the same pair/direction, the informal one was +# just the precursor announcement (e.g. "Gold sell now" immediately +# followed by the full SL/TP message) - it gets dropped so the same setup +# isn't counted as two separate trades. +INFORMAL_MERGE_WINDOW_SECONDS = 15 * 60 + +# A freshly-parsed entry that deviates more than this from the recent +# median for the same pair is treated as a likely parsing error (e.g. a +# stray number from elsewhere in the message getting picked up as the +# entry) rather than a real price, and the signal is skipped. +PRICE_SANITY_PCT = 0.12 + +# Recognized pair/asset names that may appear right before BUY/SELL +PAIR_NAME_PATTERN = r'(GOLD|XAUUSD|XAU|SILVER|XAGUSD|XAG|BITCOIN|BTCUSD|BTC|[A-Z]{6})' + +TP_DONE_RE = re.compile(r'\b(done|hit|closed)\b|✅|☑', re.IGNORECASE) +TP_NUMBER_RE = re.compile(r'\btp\s*#?\s*(\d+)\b', re.IGNORECASE) +SL_HIT_RE = re.compile(r'\bsl\s*hit\b|\bstopped\s*out\b|\bstop\s*loss\s*hit\b|\bbe\s*hit\b', re.IGNORECASE) +PARTIAL_CLOSE_RE = re.compile(r'\bclos(?:e|ing)\s+half\b|\bpartial(?:s)?\b', re.IGNORECASE) +ALL_TP_RE = re.compile(r'\ball\s+(?:take\s*profits?|tps?)\s+hit\b', re.IGNORECASE) +# Trader announcing they are moving stops to breakeven (position stays open). +# "set be", "moved to be", "sl to be", "zero risk now", "no more risk" etc. +# This is distinct from SL_HIT_RE (which means price actually hit the level). +BE_MOVE_RE = re.compile( + r'\bset\s+(?:sl\s+to\s+)?be\b' # "set be", "set sl to be" + r'|\bmov(?:e|ed|ing)\s+(?:sl\s+)?to\s+(?:be|break\s*even)\b' + r'|\bsl\s+(?:to|at)\s+(?:be|break\s*even)\b' + r'|\bzero\s+risk\b' # "zero risk now" + r'|\bno\s+more\s+risk\b', # "no more risk today" + re.IGNORECASE, +) +# Admin saying the last signal was wrong / should be ignored +CANCEL_RE = re.compile( + r'\b(?:cancel(?:led)?|wrong|mistake' + r'|ignore(?:\s+(?:last|that|this))?' + r'|don\'?t\s+take' + r'|disregard' + r'|check\s+numbers?' + r'|was\s+(?:a\s+)?(?:an?\s+)?error' + r'|bad\s+signal|false\s+signal|not\s+valid|void)\b', + re.IGNORECASE +) + + +class TradeSignalAnalyzer: + """ + Parses a Telegram signal export and runs an OBJECTIVE backtest against + real market data to see what a mechanical execution of the signals + would actually have produced - independent of what the provider claims + happened in the chat. + + The provider's own narration (e.g. "Tp1 done", "Sl hit", "set BE") is + still captured, but only as informational metadata (`reported_*` + fields) shown next to the simulated result for comparison. It is never + used as an input to the PnL math - only real OHLC price data decides + whether a TP/SL level was actually reached. + + Trade management rule that the simulation DOES bake in (because it's + described repeatedly as the provider's general method, not a one-off + claim): once price action confirms the first TP tranche, the stop for + the remaining tranches is moved to breakeven. + """ + + def __init__(self, file_path, initial_capital=100, risk_level='mid', + sl_confirmation='close', compounding=False, + default_sl_pips=60, default_tp_ladder_pips=(20, 40, 60, 80), + mt5_base_url='http://localhost:8000'): + self.file_path = file_path + self.initial_capital = initial_capital + self.risk_level = risk_level + self.sl_confirmation = sl_confirmation # 'close' (default) or 'wick' + self.compounding = compounding # False (default): lot sized off initial_capital + self.default_sl_pips = default_sl_pips + self.default_tp_ladder_pips = list(default_tp_ladder_pips) + self.mt5_base_url = mt5_base_url.rstrip('/') + self.signals = [] + self.trades_data = {} + self._recent_entries = {} # pair -> list of recently accepted entry prices + self.skipped_signals = [] # log of signals rejected by the sanity check + self.defaulted_signals = [] # log of informal signals that got default SL/TP/entry filled in + + # ------------------------------------------------------------------ + # Parsing + # ------------------------------------------------------------------ + def parse_signals(self): + with open(self.file_path, 'r', encoding='utf-8') as f: + content = f.read() + + raw_messages = re.split(r'-{10,}', content) + + parsed = [] + for raw in raw_messages: + if not raw.strip(): + continue + match = re.match(r'Date:\s*(.*?),\s*Message:\s*(.*)', raw.strip(), re.DOTALL) + if not match: + continue + date_str, msg = match.groups() + try: + date = datetime.strptime(date_str.strip(), '%Y-%m-%d %H:%M:%S%z') + except ValueError: + continue + msg = msg.strip() + if msg: + parsed.append((date, msg)) + + # Telegram exports are typically newest-first; the management + # messages (TP done, SL hit, ...) only make sense relative to the + # signal that came BEFORE them in time, so we must process in + # chronological order regardless of file order. + parsed.sort(key=lambda x: x[0]) + + for date, msg in parsed: + self.process_message(date, msg) + + self._drop_superseded_informal_signals() + + def process_message(self, date, msg): + """Handle a single message - either a new entry signal or a + management message (TP done, SL hit, partial close) against the + most recently opened signal. Used both for file replay (above) + and for a live Telegram listener processing messages one at a + time as they arrive (see live_signal_trader.py). Returns the new + signal dict if one was opened, otherwise None.""" + signal = self._parse_new_trade(msg, date) + if signal: + self.signals.append(signal) + return signal + + if not self.signals or self.signals[-1]['status'] != 'OPEN': + return None + + if CANCEL_RE.search(msg): + # Admin retracted the last signal - mark it cancelled so the + # live trader can close any position it already opened. + self.signals[-1]['status'] = 'CANCELLED' + self.signals[-1]['cancel_date'] = date + self.signals[-1]['cancel_reason'] = msg.strip() + return 'CANCEL' + + if BE_MOVE_RE.search(msg): + # Trader is moving their SL to breakeven (position stays OPEN). + # Return sentinel so the live trader can relay the SL adjustment. + self.signals[-1]['be_move_date'] = date + return 'BE_MOVE' + + if SL_HIT_RE.search(msg): + self._handle_sl_hit(date, msg) + elif ALL_TP_RE.search(msg): + self._handle_tp_report(date, msg, all_levels=True) + elif TP_NUMBER_RE.search(msg) and TP_DONE_RE.search(msg): + self._handle_tp_report(date, msg) + elif PARTIAL_CLOSE_RE.search(msg): + self._handle_partial_close(date, msg) + return None + + def _drop_superseded_informal_signals(self): + """An informal 'Gold sell now' is often just a precursor to a fully + detailed message seconds/minutes later for the same setup (exactly + like your Jan 22 sample). If a detailed signal for the same pair + and direction shows up within INFORMAL_MERGE_WINDOW after an + informal one, the informal one was only the announcement - drop it + so the trade isn't counted twice.""" + detailed = [s for s in self.signals if not s['is_informal']] + kept = [] + for s in self.signals: + if s['is_informal'] and any( + d['pair'] == s['pair'] and d['action'] == s['action'] + and 0 <= (d['date'] - s['date']).total_seconds() <= INFORMAL_MERGE_WINDOW_SECONDS + for d in detailed + ): + continue + kept.append(s) + self.signals = kept + + def _parse_new_trade(self, message, date): + """Parse an entry signal. Two shapes are recognized: + + 1. Fully detailed, e.g.: + Gold sell 4827-4831 + SL: 4835 + TP: 4825 + TP: 4823 + 2. Informal, e.g. "Gold buy now" - just a pair + direction, with no + SL/TP (sometimes no price at all). These are still returned as + tradeable signals (is_informal=True) with sl/tp_levels left as + None/[] so they can be filled in with your default risk plan + once market data is available (see _resolve_informal_signals). + """ + pair_action = re.search( + PAIR_NAME_PATTERN + r'\s+(BUY|SELL)\b', message, re.IGNORECASE + ) + if not pair_action: + return None + + pair_raw = pair_action.group(1).upper() + pair = PAIR_ALIASES.get(pair_raw, pair_raw) + action = pair_action.group(2).upper() + + # Entry: provider usually gives a range ("4827-4831"); fall back to + # a single number right after the pair/direction if no range is + # found. Restricted to the text between "BUY/SELL" and the SL + # marker - searching the whole message risked picking up an + # unrelated number (a pip count, an aside, etc.) as the entry. + # When more than one candidate range/number shows up in that + # segment (e.g. "after a 20-30 pip pullback 2640-2643"), pick + # whichever is closest to the recent price history for this pair, + # rather than blindly taking the first match. + tail = message[pair_action.end():] + entry_segment = re.split(r'\bSL\b', tail, maxsplit=1, flags=re.IGNORECASE)[0] + + range_matches = re.findall(r'(\d+(?:\.\d+)?)\s*-\s*(\d+(?:\.\d+)?)', entry_segment) + if range_matches: + candidates = [((float(a) + float(b)) / 2, float(a), float(b)) for a, b in range_matches] + else: + single_matches = re.findall(r'\d+(?:\.\d+)?', entry_segment) + candidates = [(float(x), float(x), float(x)) for x in single_matches] if single_matches else [] + + entry = entry_low = entry_high = None + if candidates: + entry, entry_low, entry_high = self._pick_best_candidate(pair, candidates) + + # SL: there should only be one real stop-loss, but a longer message + # can mention "SL" more than once (e.g. trailing commentary after + # the TP block). No fixed ordering between SL and TP is assumed - + # collect every candidate and, once the entry is known, take the + # one closest to it rather than just the first match in the text. + # The colon is required: without it, a bare "tp1"/"sl" mention in + # casual prose (no price actually following) would otherwise let + # the regex backtrack into capturing the label digit itself. + sl_candidates = [float(x) for x in + re.findall(r'\bSL\s*\d*\s*:\s*([\d]+(?:\.\d+)?)', message, re.IGNORECASE)] + if not sl_candidates: + sl = None + elif entry is not None and len(sl_candidates) > 1: + sl = min(sl_candidates, key=lambda x: abs(x - entry)) + else: + sl = sl_candidates[0] + + tp_matches = re.findall(r'\bTP\s*\d*\s*:\s*(?:open\s*)?([\d]+(?:\.\d+)?)', message, re.IGNORECASE) + + # Drop clearly truncated / corrupted TP values (e.g. "TP: 41" that should + # be "TP: 4170" but got cut off in the Telegram message) BEFORE the + # is_informal check so a single bad line doesn't cause the whole signal to + # be rejected by _is_sane. Any candidate that deviates more than + # PRICE_SANITY_PCT from the parsed entry is almost certainly a truncation. + if entry is not None and tp_matches: + tp_matches = [x for x in tp_matches + if abs(float(x) - entry) / entry <= PRICE_SANITY_PCT] + + # Direction correction: if the price structure contradicts the text keyword + # (e.g. trader typed "Gold buy" but SL is above entry → it's a SELL), trust + # the numbers over the word. This handles typos like "my mess have a error". + # Priority: SL position > TP direction > text keyword (fallback). + if entry is not None and sl is not None: + structure_action = 'SELL' if sl > entry else 'BUY' + if structure_action != action: + import logging as _log + _log.getLogger(__name__).warning( + f"Direction mismatch: text says {action} but SL {sl} is " + f"{'above' if sl > entry else 'below'} entry {entry} " + f"→ correcting to {structure_action}" + ) + action = structure_action + elif entry is not None and tp_matches: + # No SL yet - infer from TP side as a weaker signal + tp_vals = [float(x) for x in tp_matches] + tp_below = sum(1 for t in tp_vals if t < entry) + tp_above = sum(1 for t in tp_vals if t > entry) + if tp_below > tp_above and action == 'BUY': + import logging as _log + _log.getLogger(__name__).warning( + f"Direction mismatch: text says BUY but TPs are mostly below entry {entry} " + f"→ correcting to SELL" + ) + action = 'SELL' + elif tp_above > tp_below and action == 'SELL': + import logging as _log + _log.getLogger(__name__).warning( + f"Direction mismatch: text says SELL but TPs are mostly above entry {entry} " + f"→ correcting to BUY" + ) + action = 'BUY' + + is_informal = not (sl is not None and tp_matches) + + if not is_informal: + # Fully detailed signal: apply the strict checks, since a bad + # parse here silently corrupts the simulation (as you saw). + if entry is None: + return None + tp_levels = sorted({float(x) for x in tp_matches}, reverse=(action == 'SELL')) + if not self._is_sane(pair, action, entry, sl, tp_levels, date): + return None + self._recent_entries.setdefault(pair, []).append(entry) + else: + # Informal signal: SL/TP will be filled with your default risk + # plan later. An explicit entry price is only kept if it's not + # an obvious mis-parse; otherwise it's left None and resolved + # from market data (the next available price) once data is + # fetched, rather than rejecting the whole signal. + tp_levels = sorted({float(x) for x in tp_matches}, reverse=(action == 'SELL')) if tp_matches else [] + if entry is not None and not self._entry_plausible(pair, entry): + entry = entry_low = entry_high = None + if entry is not None: + self._recent_entries.setdefault(pair, []).append(entry) + + return { + 'date': date, + 'pair': pair, + 'pair_raw': pair_raw, + 'action': action, + 'entry': entry, # may be None for informal signals - resolved from market data later + 'entry_low': entry_low, + 'entry_high': entry_high, + 'sl': sl, # may be None for informal signals - default-filled later + 'tp_levels': tp_levels, # may be [] for informal signals - default-filled later + 'is_informal': is_informal, + 'status': 'OPEN', + # Informational only - filled in by management messages below. + 'reported_tp_hits': set(), + 'reported_close_type': None, + 'reported_close_date': None, + } + + def _pick_best_candidate(self, pair, candidates): + """candidates: list of (entry, low, high) tuples. Picks whichever + entry is closest to the recent price history for this pair; with + no history yet, falls back to the last candidate in the segment + (in practice the actual price tends to be stated right before the + SL line, after any incidental pip/context mentions).""" + recent = self._recent_entries.get(pair, []) + if recent: + reference = statistics.median(recent[-5:]) + return min(candidates, key=lambda c: abs(c[0] - reference)) + return candidates[-1] + + def _entry_plausible(self, pair, entry): + recent = self._recent_entries.get(pair, []) + if len(recent) >= 3: + reference = statistics.median(recent[-5:]) + return abs(entry - reference) / reference <= PRICE_SANITY_PCT + return True + + def _is_sane(self, pair, action, entry, sl, tp_levels, date): + """Reject a parsed signal if the entry looks like a mis-parse + rather than a real price (e.g. entry=3141.75 / 1351.00 showing up + among neighbors all around 2650 - a stray number got captured + instead of the actual entry), if SL/TP sit implausibly far from + the entry for what's meant to be a scalp setup, or if a TP/SL + ended up on the wrong side of entry for the trade direction (a + SELL's TP must be below entry, its SL above - otherwise "hitting + the TP" mechanically realizes a loss instead of a profit, which + is exactly how one trade in your run wiped the whole account on a + single signal despite showing "all TPs hit"). + """ + recent = self._recent_entries.get(pair, []) + if len(recent) >= 3: + reference = statistics.median(recent[-5:]) + if abs(entry - reference) / reference > PRICE_SANITY_PCT: + self.skipped_signals.append((date, pair, entry, 'entry deviates from recent median')) + print(f"Skipping signal at {date} ({pair}): entry {entry} is " + f">{PRICE_SANITY_PCT*100:.0f}% away from the recent median " + f"({reference:.2f}) - looks like a parsing error, check the source message.") + return False + + for label, level in [('SL', sl)] + [(f'TP{i+1}', t) for i, t in enumerate(tp_levels)]: + if abs(level - entry) / entry > PRICE_SANITY_PCT: + self.skipped_signals.append((date, pair, entry, f'{label} too far from entry')) + print(f"Skipping signal at {date} ({pair}): {label}={level} is implausibly far " + f"from entry {entry} - looks like a parsing error, check the source message.") + return False + + sl_wrong_side = (action == 'BUY' and sl >= entry) or (action == 'SELL' and sl <= entry) + if sl_wrong_side: + self.skipped_signals.append((date, pair, entry, 'SL on the wrong side of entry')) + print(f"Skipping signal at {date} ({pair}): SL={sl} is on the wrong side of " + f"entry {entry} for a {action} - looks like a parsing error, check the source message.") + return False + + tp_wrong_side = any((action == 'BUY' and tp <= entry) or (action == 'SELL' and tp >= entry) + for tp in tp_levels) + if tp_wrong_side: + self.skipped_signals.append((date, pair, entry, 'a TP is on the wrong side of entry')) + print(f"Skipping signal at {date} ({pair}): one of {tp_levels} is on the wrong side of " + f"entry {entry} for a {action} - looks like a parsing error, check the source message.") + return False + + return True + + def _handle_sl_hit(self, date, msg): + sig = self.signals[-1] + sig['status'] = 'CLOSED' + sig['reported_close_type'] = 'BE' if 'be hit' in msg.lower() else 'SL' + sig['reported_close_date'] = date + + def _handle_tp_report(self, date, msg, all_levels=False): + sig = self.signals[-1] + if all_levels: + sig['reported_tp_hits'].update(range(len(sig['tp_levels']))) + else: + for n in TP_NUMBER_RE.findall(msg): + idx = int(n) - 1 + if 0 <= idx < len(sig['tp_levels']): + sig['reported_tp_hits'].add(idx) + if len(sig['reported_tp_hits']) >= len(sig['tp_levels']) and sig['tp_levels']: + sig['status'] = 'CLOSED' + sig['reported_close_type'] = 'ALL_TP' + sig['reported_close_date'] = date + + def _handle_partial_close(self, date, msg): + sig = self.signals[-1] + sig['reported_partial_close_date'] = date + + # ------------------------------------------------------------------ + # Market data + # ------------------------------------------------------------------ + def fetch_forex_data(self): + unique_pairs = {s['pair'] for s in self.signals} + + for pair in unique_pairs: + symbol = MT5_SYMBOL_OVERRIDES.get(pair, pair) + dates = [s['date'].replace(tzinfo=None) for s in self.signals if s['pair'] == pair] + earliest_needed = min(dates) + + df = None + for tf in MT5_TIMEFRAME_FALLBACK_CHAIN: + candidate = self._mt5_fetch_ohlc(symbol, tf) + if candidate is None or candidate.empty: + continue + if candidate.index.min() <= earliest_needed: + df = candidate + print(f"{pair} ({symbol}): using {tf} bars, " + f"{candidate.index.min()} -> {candidate.index.max()} ({len(candidate)} bars)") + break + print(f"{pair} ({symbol}): {tf} only reaches back to {candidate.index.min()}, " + f"need {earliest_needed} - trying a coarser timeframe") + df = candidate # keep the coarsest successful fetch as a fallback + + if df is None: + print(f"No MT5 data available for {pair} ({symbol}) - check the symbol name " + f"via GET {self.mt5_base_url}/symbols") + self.trades_data[pair] = df + + self._resolve_informal_signals() + + def _resolve_informal_signals(self): + """Fill in whatever an informal signal ('Gold buy now') didn't + specify: entry, if no price was in the message, comes from the + first available market price at/after the signal time (a market- + order fill); SL/TP, if missing, come from your default risk plan + (default_sl_pips / default_tp_ladder_pips), centered on whichever + entry ended up being used (explicit or market-resolved).""" + for sig in self.signals: + if not sig['is_informal']: + continue + + data = self.trades_data.get(sig['pair']) + + if sig['entry'] is None: + if data is None or data.empty: + continue # no entry and no data - will show as NO_DATA downstream + signal_date = sig['date'].replace(tzinfo=None) + future = data[data.index >= signal_date] + if future.empty: + continue + sig['entry'] = float(future.iloc[0]['Open']) + sig['entry_low'] = sig['entry_high'] = sig['entry'] + self._recent_entries.setdefault(sig['pair'], []).append(sig['entry']) + self.defaulted_signals.append( + (sig['date'], sig['pair'], sig['action'], 'entry', + f"no price in message, used market open {sig['entry']:.2f}")) + + pip = PIP_SIZE.get(sig['pair'], DEFAULT_PIP_SIZE) + + if sig['sl'] is None: + distance = self.default_sl_pips * pip + sig['sl'] = sig['entry'] - distance if sig['action'] == 'BUY' else sig['entry'] + distance + self.defaulted_signals.append( + (sig['date'], sig['pair'], sig['action'], 'SL', + f"no SL in message, defaulted to {self.default_sl_pips} pips ({sig['sl']:.2f})")) + + if not sig['tp_levels']: + levels = [sig['entry'] + (d * pip if sig['action'] == 'BUY' else -d * pip) + for d in self.default_tp_ladder_pips] + sig['tp_levels'] = sorted(levels, reverse=(sig['action'] == 'SELL')) + self.defaulted_signals.append( + (sig['date'], sig['pair'], sig['action'], 'TP', + f"no TP in message, defaulted to {self.default_tp_ladder_pips} pip ladder")) + + def _mt5_fetch_ohlc(self, symbol, timeframe, bars=10000): + """Pull OHLC bars from your MT5 MCP server's /ohlc endpoint.""" + url = f"{self.mt5_base_url}/ohlc/{symbol}" + try: + resp = requests.get(url, params={'timeframe': timeframe, 'bars': bars}, timeout=30) + resp.raise_for_status() + payload = resp.json() + except Exception as e: + print(f"MT5 MCP fetch failed for {symbol} {timeframe}: {e}") + return None + + rows = payload.get('data', []) + if not rows: + return None + + df = pd.DataFrame(rows) + df['time'] = pd.to_datetime(df['time'], utc=True).dt.tz_localize(None) + df = df.set_index('time').sort_index() + df = df.rename(columns={ + 'open': 'Open', 'high': 'High', 'low': 'Low', + 'close': 'Close', 'volume': 'Volume', + }) + return df + + # ------------------------------------------------------------------ + # Backtest + # ------------------------------------------------------------------ + def analyze_trades(self, risk_level=None, sl_confirmation=None, compounding=None, + sizing_mode='fixed', min_lot=MIN_LOT): + """Run a bar-by-bar backtest over all parsed signals. + + sizing_mode options: + 'fixed' - lot = (capital / 1000) * RISK_TIERS[risk_level] + 'risk_pct' - lot sized so a full stop-out costs exactly + RISK_PCT_TIERS[risk_level] of capital + 'survival' - lot sized so THREE full stop-outs would exhaust the + current capital (risk_pct = 1/3). If the resulting + lot < min_lot, the signal is traded with min_lot at + TP1 only (the same "small-account" fallback the live + trader uses). This is the most realistic mode for a + $100 starting balance. + """ + risk_level = risk_level or self.risk_level + sl_confirmation = sl_confirmation or self.sl_confirmation + compounding = self.compounding if compounding is None else compounding + results = [] + current_capital = self.initial_capital + max_capital = current_capital + max_drawdown = 0.0 + position_open_until = None # bar timestamp when the last trade closed + + for signal in self.signals: + pair = signal['pair'] + data = self.trades_data.get(pair) + + row = { + 'date': signal['date'], + 'pair': pair, + 'action': signal['action'], + 'entry': signal['entry'], + 'sl': signal['sl'], + 'tp_count': len(signal['tp_levels']), + 'signal_type': 'informal' if signal['is_informal'] else 'detailed', + 'reported_result': self._reported_result_label(signal), + } + + if signal['entry'] is None or data is None or data.empty: + row.update(result='NO_DATA', pnl=0.0, pnl_pct=0.0, capital=current_capital, + drawdown=max_drawdown, sizing='NO_DATA') + results.append(row) + continue + + signal_date = signal['date'].replace(tzinfo=None) + + # Skip signals that arrive while a previous trade is still open — + # mirrors the live trader's max_concurrent=1 guard exactly. + if position_open_until is not None and signal_date <= position_open_until: + row.update(result='SKIPPED(concurrent)', pnl=0.0, pnl_pct=0.0, + capital=current_capital, drawdown=max_drawdown, lot=0.0, + sizing='blocked') + results.append(row) + continue + + trade_data = data[data.index >= signal_date] + if trade_data.empty: + row.update(result='NO_DATA', pnl=0.0, pnl_pct=0.0, capital=current_capital, + drawdown=max_drawdown, sizing='NO_DATA') + results.append(row) + continue + + sizing_capital = current_capital if compounding else self.initial_capital + + if sizing_mode == 'survival': + # 3-signal survival rule (mirrors the live trader exactly): + # lot that costs balance/3 on a full stop-out + survival_lot = risk_based_lot(sizing_capital, 1.0 / 3.0, + signal['entry'], signal['sl'], pair) + if survival_lot < min_lot: + # Too small for the full ladder - use min_lot / TP1 only + lot = min_lot + effective_tp_levels = signal['tp_levels'][:1] + sim_signal = dict(signal, tp_levels=effective_tp_levels) + sizing_tag = f'survival/TP1({min_lot})' + else: + lot = survival_lot + sim_signal = signal + sizing_tag = f'survival({lot:.4f})' + elif sizing_mode == 'risk_pct': + lot = risk_based_lot(sizing_capital, RISK_PCT_TIERS[risk_level], + signal['entry'], signal['sl'], pair) + sim_signal = signal + sizing_tag = f'risk_pct/{risk_level}({lot:.4f})' + else: + lot = (sizing_capital / 1000.0) * RISK_TIERS[risk_level] + sim_signal = signal + sizing_tag = f'fixed/{risk_level}({lot:.4f})' + + pnl, sim_result, close_time = self._simulate_trade(sim_signal, trade_data, lot, sl_confirmation) + + # Record when this position closed so subsequent signals can be blocked. + if close_time is not None: + position_open_until = close_time + + capital_before = current_capital + current_capital = max(0.0, current_capital + pnl) + max_capital = max(max_capital, current_capital) + drawdown = (max_capital - current_capital) / max_capital * 100 if max_capital > 0 else 0.0 + max_drawdown = max(max_drawdown, drawdown) + pnl_pct = (pnl / capital_before * 100) if capital_before else 0.0 + + row.update(result=sim_result, pnl=pnl, pnl_pct=pnl_pct, capital=current_capital, + drawdown=drawdown, lot=lot, sizing=sizing_tag) + results.append(row) + + return pd.DataFrame(results) + + def _simulate_trade(self, signal, trade_data, lot, sl_confirmation): + """Bar-by-bar simulation against real OHLC data. + + Position size is FIXED at trade-open time: lot (sized off current + equity via your risk-tier table) x the instrument's contract size, + split evenly across one tranche per TP level. This replaces an + earlier leverage*equity model that re-leveraged the whole, already + compounded balance into every trade and blew up exponentially. + + TP fills use the bar's High/Low (a touch is a real limit fill). + SL/breakeven exits use the bar's Close by default (sl_confirmation + ='close') rather than the wick, mirroring your stated approach of + not hard-placing the stop so a brief spread spike doesn't take you + out. Pass sl_confirmation='wick' for the stricter, literal version. + """ + action = signal['action'] + entry_price = signal['entry'] + tp_levels = signal['tp_levels'] + n_tranches = max(len(tp_levels), 1) + + contract_size = CONTRACT_SIZE.get(signal['pair'], DEFAULT_CONTRACT_SIZE) + dollar_per_unit = lot * contract_size + tranche_dollar_per_unit = dollar_per_unit / n_tranches + + current_sl = signal['sl'] + remaining = list(range(len(tp_levels))) + pnl = 0.0 + hit_labels = [] + + close_bar_time = None + for bar_time, bar in trade_data.iterrows(): + close_bar_time = bar_time + high, low, close = float(bar['High']), float(bar['Low']), float(bar['Close']) + sl_ref = close if sl_confirmation == 'close' else (low if action == 'BUY' else high) + + sl_triggered = (action == 'BUY' and sl_ref <= current_sl) or \ + (action == 'SELL' and sl_ref >= current_sl) + if sl_triggered and remaining: + for _ in remaining: + if action == 'BUY': + pnl += tranche_dollar_per_unit * (current_sl - entry_price) + else: + pnl += tranche_dollar_per_unit * (entry_price - current_sl) + hit_labels.append('BE' if current_sl == entry_price else 'SL') + remaining = [] + break + + for idx in list(remaining): + tp_price = tp_levels[idx] + hit = (action == 'BUY' and high >= tp_price) or \ + (action == 'SELL' and low <= tp_price) + if hit: + if action == 'BUY': + pnl += tranche_dollar_per_unit * (tp_price - entry_price) + else: + pnl += tranche_dollar_per_unit * (entry_price - tp_price) + remaining.remove(idx) + hit_labels.append(f'TP{idx + 1}') + if current_sl != entry_price: + current_sl = entry_price # breakeven after first confirmed TP + + if not remaining: + break + + if remaining: + last_close = float(trade_data.iloc[-1]['Close']) + for _ in remaining: + if action == 'BUY': + pnl += tranche_dollar_per_unit * (last_close - entry_price) + else: + pnl += tranche_dollar_per_unit * (entry_price - last_close) + hit_labels.append('OPEN(MTM)') + + return pnl, '+'.join(hit_labels) if hit_labels else 'UNKNOWN', close_bar_time + + @staticmethod + def _reported_result_label(signal): + if signal['reported_close_type'] == 'ALL_TP': + return f"ALL_TP({len(signal['tp_levels'])})" + if signal['reported_close_type']: + tps = ','.join(f'TP{i+1}' for i in sorted(signal['reported_tp_hits'])) + return f"{tps}+{signal['reported_close_type']}" if tps else signal['reported_close_type'] + if signal['reported_tp_hits']: + return ','.join(f'TP{i+1}' for i in sorted(signal['reported_tp_hits'])) + return 'OPEN' + + +def _fetch_from_telegram(analyzer, channel_id, days): + """Pull `days` days of history from a Telegram channel and feed each + message through analyzer.process_message(). Returns (total_msgs, skipped). + + Requires TELEGRAM_API_ID / TELEGRAM_API_HASH / PHONE_NUMBER env vars. + Uses a temporary session file so it never conflicts with the live trader. + """ + import os + from dotenv import load_dotenv + load_dotenv() + from datetime import datetime, timezone, timedelta + from telethon.sync import TelegramClient + + api_id = int(os.environ['API_ID']) + api_hash = os.environ['API_HASH'] + phone = os.environ['PHONE_NUMBER'] + + cutoff = datetime.now(timezone.utc) - timedelta(days=days) + print(f"Connecting to Telegram, fetching {days} days of history " + f"(since {cutoff.date()}) from channel {channel_id} …") + + total_msgs = 0 + skipped = 0 + # Reuse the live trader's existing session so no re-authentication is needed. + session_name = 'mysession' + + with TelegramClient(session_name, api_id, api_hash) as client: + client.start(phone=phone) + + # Resolve the channel by walking iter_dialogs() — this gives us the + # fully resolved entity (with access hash) regardless of session cache. + bare_id = abs(channel_id) + if str(bare_id).startswith('100') and len(str(bare_id)) > 10: + bare_id = int(str(bare_id)[3:]) + + entity = None + print("Resolving channel from dialogs…") + for dialog in client.iter_dialogs(): + if getattr(dialog.entity, 'id', None) == bare_id: + entity = dialog.entity + break + + if entity is None: + raise ValueError( + f"Channel {channel_id} not found in your dialogs. " + "Make sure you are a member of the channel." + ) + + for msg in client.iter_messages(entity, reverse=True, offset_date=cutoff): + text = (msg.text or '').strip() + if not text: + skipped += 1 + continue + total_msgs += 1 + analyzer.process_message(msg.date, text) + + analyzer._drop_superseded_informal_signals() + return total_msgs, skipped + + +def main(): + """Standalone backtester for the Goldex channel export. + + File mode (default): + python3 traderanalyser.py [mt5_url] [tier] [compound] [capital] [sizing] + + Telegram mode (live history fetch – no file needed): + python3 traderanalyser.py --telegram [channel_id] [days] [mt5_url] [tier] [compound] [capital] [sizing] + + In Telegram mode the script connects using TELEGRAM_API_ID / TELEGRAM_API_HASH / + PHONE_NUMBER env vars, fetches the last N days (default 5) of history from + the Goldex channel, feeds every message through the parser, then runs the + full backtest — identical output to file mode. + + channel_id defaults to the CHANNEL_ID env var. + """ + import sys + import os + + # ── Telegram mode ──────────────────────────────────────────────────────── + if len(sys.argv) > 1 and sys.argv[1] == '--telegram': + from dotenv import load_dotenv + load_dotenv() + argv = sys.argv[2:] + channel_id = int(argv[0]) if len(argv) > 0 else int(os.environ.get('CHANNEL_ID', 0)) + days = int(argv[1]) if len(argv) > 1 else 5 + mt5_base_url = argv[2] if len(argv) > 2 else 'http://localhost:8000' + chosen_tier = argv[3] if len(argv) > 3 else 'mid' + compounding = argv[4].lower() in ('1', 'true', 'yes', 'compound') if len(argv) > 4 else False + initial_capital = float(argv[5]) if len(argv) > 5 else 100 + sizing_mode = argv[6] if len(argv) > 6 else 'survival' + + if not channel_id: + print("ERROR: channel_id required (pass as arg or set CHANNEL_ID env var).") + sys.exit(1) + + analyzer = TradeSignalAnalyzer( + file_path='', + initial_capital=initial_capital, + compounding=compounding, + mt5_base_url=mt5_base_url, + ) + total_msgs, skipped = _fetch_from_telegram(analyzer, channel_id, days) + print(f"Fetched {total_msgs} messages ({skipped} empty/media skipped) " + f"from channel {channel_id} over the last {days} days.") + + # ── File mode ──────────────────────────────────────────────────────────── + else: + file_path = sys.argv[1] if len(sys.argv) > 1 else 'telegram_messages.txt' + mt5_base_url = sys.argv[2] if len(sys.argv) > 2 else 'http://localhost:8000' + chosen_tier = sys.argv[3] if len(sys.argv) > 3 else 'mid' + compounding = (sys.argv[4].lower() in ('1', 'true', 'yes', 'compound') + if len(sys.argv) > 4 else False) + initial_capital = float(sys.argv[5]) if len(sys.argv) > 5 else 100 + sizing_mode = sys.argv[6] if len(sys.argv) > 6 else 'survival' + + analyzer = TradeSignalAnalyzer( + file_path, initial_capital=initial_capital, + compounding=compounding, mt5_base_url=mt5_base_url, + ) + analyzer.parse_signals() + + if not analyzer.signals: + print("No complete trade signals (pair + direction + SL + TP) were found.") + return + + if analyzer.skipped_signals: + print(f"\n{len(analyzer.skipped_signals)} signal(s) skipped by the sanity check " + f"(likely parsing errors - worth checking the source messages):") + for date, pair, entry, reason in analyzer.skipped_signals: + print(f" {date} {pair} entry={entry} ({reason})") + + n_informal = sum(1 for s in analyzer.signals if s['is_informal']) + if n_informal: + print(f"\n{n_informal} informal signal(s) (e.g. \"Gold buy now\") kept for trading " + f"with defaults filled in:") + + analyzer.fetch_forex_data() + + if analyzer.defaulted_signals: + for date, pair, action, field, note in analyzer.defaulted_signals: + print(f" {date} {pair} {action}: {note}") + + mode_label = 'COMPOUNDING (lot grows with balance)' if compounding else 'FIXED (lot sized off initial capital only)' + print(f"\nRisk tier comparison [{mode_label}] " + f"- pass 'compound' as a 4th argument to switch modes:") + print("-" * 70) + print(" FIXED table (lot per $1000 of capital - ignores stop width):") + for tier in ('high', 'mid', 'low'): + df = analyzer.analyze_trades(risk_level=tier, sizing_mode='fixed') + evaluated = df[df['result'] != 'NO_DATA'] + final_capital = df['capital'].iloc[-1] if not df.empty else analyzer.initial_capital + total_return = (final_capital - analyzer.initial_capital) / analyzer.initial_capital * 100 + win_rate = evaluated['pnl'].gt(0).mean() * 100 if not evaluated.empty else 0.0 + print(f" {tier:>5} (lot={RISK_TIERS[tier]:.2f}/$1000): " + f"final capital {final_capital:,.2f} USDT, return {total_return:+.1f}%, " + f"max drawdown {df['drawdown'].max():.1f}%, win rate {win_rate:.1f}%") + + print(" RISK_PCT table (risk % of capital per trade, scaled to each trade's actual stop):") + risk_pct_results = {} + for tier in ('high', 'mid', 'low'): + df = analyzer.analyze_trades(risk_level=tier, sizing_mode='risk_pct') + risk_pct_results[tier] = df + evaluated = df[df['result'] != 'NO_DATA'] + final_capital = df['capital'].iloc[-1] if not df.empty else analyzer.initial_capital + total_return = (final_capital - analyzer.initial_capital) / analyzer.initial_capital * 100 + win_rate = evaluated['pnl'].gt(0).mean() * 100 if not evaluated.empty else 0.0 + print(f" {tier:>5} (risk={RISK_PCT_TIERS[tier]*100:.1f}%/trade): " + f"final capital {final_capital:,.2f} USDT, return {total_return:+.1f}%, " + f"max drawdown {df['drawdown'].max():.1f}%, win rate {win_rate:.1f}%") + + print(" SURVIVAL table (risk 1/3 of balance per trade; if survival lot < 0.01 → TP1 only):") + df_surv = analyzer.analyze_trades(sizing_mode='survival', compounding=True) + evaluated_surv = df_surv[df_surv['result'] != 'NO_DATA'] + final_surv = df_surv['capital'].iloc[-1] if not df_surv.empty else analyzer.initial_capital + total_return_surv = (final_surv - analyzer.initial_capital) / analyzer.initial_capital * 100 + win_rate_surv = evaluated_surv['pnl'].gt(0).mean() * 100 if not evaluated_surv.empty else 0.0 + n_survival_mode = (df_surv['sizing'].str.startswith('survival/TP1')).sum() if 'sizing' in df_surv.columns else '?' + print(f" (compounding, 1/3 risk): " + f"final capital {final_surv:,.2f} USDT, return {total_return_surv:+.1f}%, " + f"max drawdown {df_surv['drawdown'].max():.1f}%, win rate {win_rate_surv:.1f}% " + f"({n_survival_mode} trades in TP1-only fallback mode)") + + results_df = analyzer.analyze_trades(risk_level=chosen_tier, sizing_mode=sizing_mode, + compounding=compounding) + + if results_df.empty or 'capital' not in results_df.columns: + print("No trades could be evaluated (no signals or no market data available).") + return + + tier_desc = '' if sizing_mode == 'survival' else f"'{chosen_tier}' tier, " + print(f"\nDetailed results below use {tier_desc}'{sizing_mode}' sizing " + f"(pass high/mid/low as 3rd arg, 'fixed'/'risk_pct'/'survival' as 6th, to switch):") + print("-" * 50) + print(f"Initial Capital: {analyzer.initial_capital} USDT") + print(f"Final Capital: {results_df['capital'].iloc[-1]:.2f} USDT") + print(f"Total Return: {((results_df['capital'].iloc[-1] - analyzer.initial_capital) / analyzer.initial_capital * 100):.2f}%") + print(f"Maximum Drawdown: {results_df['drawdown'].max():.2f}%") + + evaluated = results_df[results_df['result'] != 'NO_DATA'] + print(f"Total Signals: {len(results_df)} (No data: {(results_df['result'] == 'NO_DATA').sum()})") + if not evaluated.empty: + win_rate = evaluated['pnl'].gt(0).mean() * 100 + print(f"Win Rate (simulated): {win_rate:.2f}%") + print(f"Best trade: {evaluated['pnl'].max():.2f} USDT") + print(f"Worst trade: {evaluated['pnl'].min():.2f} USDT") + print(f"Average pnl/trade: {evaluated['pnl'].mean():.2f} USDT") + if sizing_mode == 'survival' and 'sizing' in results_df.columns: + n_tp1 = results_df['sizing'].str.startswith('survival/TP1').sum() + print(f"Trades in TP1-only fallback (balance too small for ladder): {n_tp1}") + + results_df.to_csv('trading_results.csv', index=False) + print("\nFull per-trade results saved to trading_results.csv") + + print("\nDetailed Trade Results (per-trade profit in 'pnl' / 'pnl_pct'):") + base_cols = ['date', 'pair', 'action', 'entry', 'signal_type', 'lot', 'reported_result', 'result', 'pnl', 'pnl_pct', 'capital'] + extra_cols = ['sizing'] if sizing_mode == 'survival' and 'sizing' in results_df.columns else [] + cols = base_cols + extra_cols + display_df = results_df[[c for c in cols if c in results_df.columns]].copy() + if 'lot' in display_df.columns: + display_df['lot'] = display_df['lot'].round(4) + display_df['pnl'] = display_df['pnl'].round(2) + display_df['pnl_pct'] = display_df['pnl_pct'].round(2) + display_df['capital'] = display_df['capital'].round(2) + print(display_df.to_string()) + + +if __name__ == "__main__": + main() diff --git a/docker-compose.yaml b/docker-compose.yaml index 00a0d427..6c3c9d9e 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -1,13 +1,64 @@ services: + # 1. MetaTrader 5 (Wine + KasmVNC) – startet selbst den RPyC-Server auf 8001 mt5: - build: . - image: gmag11/metatrader5_vnc + image: gmag11/metatrader5_vnc # kein "build: ." – fertiges Image reicht container_name: mt5 volumes: - ./config:/config ports: - - 3000:3000 - - 8001:8001 + - 3000:3000 # Web-UI + # 8001 (RPyC) bewusst NICHT veröffentlicht – ungeschützter Vollzugriff auf MT5 env_file: - - .env + - .env # CUSTOM_USER / PASSWORD fürs Web-UI + restart: unless-stopped + healthcheck: + test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/127.0.0.1/8001"] + interval: 30s + timeout: 5s + start_period: 10m # Erststart installiert Wine + MT5 + retries: 5 + # 2. MCP-Server (FastAPI) – spricht per RPyC mit MT5, Broker-Login aus bots.env + mt5-mcp: + build: ./bots + image: mt5-bots:local + container_name: mt5-mcp + command: ["python", "mcp_server.py", "--live", "--rpyc-host", "mt5", "--rpyc-port", "8001"] + env_file: + - bots.env + ports: + - 127.0.0.1:8000:8000 # nur lokal – hat Order-Endpunkte ohne Auth + depends_on: + mt5: + condition: service_healthy + restart: unless-stopped + healthcheck: + test: ["CMD", "python", "-c", + "import json,urllib.request,sys; d=json.load(urllib.request.urlopen('http://127.0.0.1:8000/health',timeout=5)); sys.exit(0 if d.get('connected') else 1)"] + interval: 30s + timeout: 10s + start_period: 2m + retries: 3 + + # 3. Telegram Live-Trader – handelt über den MCP-Server + live-trader: + image: mt5-bots:local + container_name: live-trader + working_dir: /data # Session + diagnostics_*.json landen im Volume + # Nur Kanal 2 (HM XAUUSD) – CHANNEL2_ID kommt aus bots.env + command: > + python /app/live_signal_trader.py --live + --channels ch2 + --risk-tier low + --mt5-base-url http://mt5-mcp:8000 + --diagnostics-dir /data + env_file: + - bots.env + environment: + - TG_SESSION=/data/mysession + volumes: + - ./trader-data:/data + depends_on: + mt5-mcp: + condition: service_healthy + restart: unless-stopped