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()