From ea2e37f0a71e98f6557befb2f50cdb0df2f44108 Mon Sep 17 00:00:00 2001 From: bast Date: Wed, 4 Nov 2026 22:58:59 +0100 Subject: [PATCH] hm s p --- bots/hm_signal_parser.py | 446 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 446 insertions(+) create mode 100644 bots/hm_signal_parser.py diff --git a/bots/hm_signal_parser.py b/bots/hm_signal_parser.py new file mode 100644 index 00000000..8102c3c1 --- /dev/null +++ b/bots/hm_signal_parser.py @@ -0,0 +1,446 @@ +""" +Parser for the HM XAUUSD Telegram channel (channel 2). + +Message format (raw Telegram Web copy-paste): + + HM XAUUSD: + SIGNAL ALERT + + SELL XAUUSD 4145-4155 + + TP1: 4140 + TP2: 4130 + ... + SL: 4165 + + HM XAUUSD: + TP1 hit, that's +70 pips + + HM XAUUSD: + BIN AUF BREAK EVEN !!! + +Key differences from channel 1 (Goldex): + - Messages are separated by the "HM XAUUSD:" sender prefix, not by dash lines + - No Date: / Message: wrapper in the raw copy-paste format + β†’ dates are assigned sequentially from an optional start timestamp, + or left as None when used purely from the live Telethon listener + (which always supplies real timestamps) + - "SIGNAL ALERT" header line precedes the actual signal body + - TP done: "TP1 hit, that's +N pips" (existing TP_DONE_RE already catches "hit") + - SL hit: "SL hit, that's -N pips" (existing SL_HIT_RE already catches this) + - BE: "BIN AUF BREAK EVEN" / "BIN AUF BE" (German) - NOT in base class + - Risk label at end of signal: "MID RISK" / "HIGH RISK" - stored as metadata + - REENTRY messages: skipped (treated as chatter, not a new signal) + - Informal: bare "SELL XAUUSD" / "BUY XAUUSD" with no price +""" +import re +from datetime import datetime, timezone, timedelta + +from traderanalyser import ( + TradeSignalAnalyzer, + TP_DONE_RE, TP_NUMBER_RE, ALL_TP_RE, PARTIAL_CLOSE_RE, CANCEL_RE, BE_MOVE_RE, +) + +# Extend the base SL_HIT regex to include German break-even phrases. +# The base class catches "sl hit", "stopped out", "stop loss hit", "be hit". +# HM uses "BIN AUF BREAK EVEN", "BIN AUF BE", and "BIN ALLE AUF BE RAUS" +# (= "I've exited all at breakeven"). The ALLE/RAUS variants insert words +# between BIN and AUF/BE, so we can't rely on word-adjacency alone. +HM_SL_HIT_RE = re.compile( + r'\bsl\s*hit\b' + r'|\bstopped\s*out\b' + r'|\bstop\s*loss\s*hit\b' + r'|\bbe\s*hit\b' + r'|bin\s+auf\s+break\s*even' # BIN AUF BREAK EVEN + r'|bin\s+auf\s+be\b' # BIN AUF BE + r'|bin\s+\w+\s+auf\s+be\b' # BIN ALLE AUF BE (one word between) + r'|auf\s+be\s+raus' # AUF BE RAUS (exited at BE) + r'|alle\s+(?:\w+\s+)*auf\s+be\b' # ALLE ... AUF BE + r'|sl\s+auf\s+(?:be|break\s*even)\b' # SL AUF BE / SL AUF BREAK EVEN + r'|jetzt\s+auf\s+be\b', # JETZT AUF BE (now to BE) + re.IGNORECASE, +) + +# "REENTRY" lines are channel-specific chatter - not new signals, not +# management messages. We skip them by detecting the keyword early. +REENTRY_RE = re.compile(r'\breentry\b', re.IGNORECASE) + +# "RUNNING X pips" - informational, skip +RUNNING_RE = re.compile(r'\brunning\b.*\bpips\b', re.IGNORECASE) + +# Risk label at the end of a signal block +RISK_LABEL_RE = re.compile(r'\b(high|mid|low)\s+risk\b', re.IGNORECASE) + + +class HMTradeSignalAnalyzer(TradeSignalAnalyzer): + """Subclass of TradeSignalAnalyzer wired to the HM XAUUSD channel format. + + Overrides only what differs: the file-parsing entry point (messages + delimited by "HM XAUUSD:" prefixes rather than dash separators, no Date + field), BE detection (German phrases), and REENTRY/RUNNING skipping. + Everything else - entry/SL/TP extraction, sanity checks, simulation, + analysis - is inherited unchanged. + """ + + def parse_signals(self, start_date=None): + """Parse a raw Telegram Web copy-paste file. + + Because the copy-paste format carries no timestamps, synthetic dates + are assigned starting from `start_date` (default: 2026-01-01 UTC), + incrementing by 1 minute per message so the chronological sort in the + base class works correctly. When used from the live Telethon listener + (process_message called directly), real timestamps are always supplied + by Telethon and this method is never called. + """ + if start_date is None: + start_date = datetime(2026, 1, 1, tzinfo=timezone.utc) + + with open(self.file_path, 'r', encoding='utf-8') as f: + content = f.read() + + # Split on "HM XAUUSD:" prefix (the sender header Telegram Web inserts + # before each message). The pattern matches the prefix at the start of + # a line, optionally followed by a newline before the message body. + parts = re.split(r'(?m)^HM XAUUSD:\s*\n?', content) + + parsed = [] + for i, part in enumerate(parts): + msg = part.strip() + if not msg: + continue + # Skip sticker / photo placeholders + if re.match(r'^(πŸ˜‘\s*Sticker|πŸ–Ό\s*Photo)$', msg, re.IGNORECASE): + continue + date = start_date + timedelta(minutes=i) + parsed.append((date, msg)) + + # Already in order (copy-paste is chronological), but sort anyway for + # consistency with the base class contract. + 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): + """Same as base class but using the extended HM regexes.""" + # REENTRY / RUNNING lines are chatter - skip before any other check + if REENTRY_RE.search(msg) or RUNNING_RE.search(msg): + return None + + # Extract optional risk label (stored on the signal for reference) + risk_match = RISK_LABEL_RE.search(msg) + risk_hint = risk_match.group(1).lower() if risk_match else None + + # Strip the "SIGNAL ALERT" header line - it's just a preamble and + # confuses the entry-range extraction if left in. + msg_clean = re.sub(r'(?im)^\s*SIGNAL ALERT\s*$', '', msg).strip() + + # HM uses "SELL XAUUSD 4145-4155" (action before pair). + # The base parser expects "XAUUSD SELL ..." so swap the order here. + msg_clean = re.sub( + r'(?i)\b(BUY|SELL)\s+(XAUUSD|GOLD|XAU|BTCUSD|BTC|XAGUSD|SILVER)\b', + r'\2 \1', + msg_clean, + ) + + signal = self._parse_new_trade(msg_clean, date) + if signal: + if risk_hint: + signal['risk_hint'] = risk_hint # informational metadata + self.signals.append(signal) + return signal + + if not self.signals or self.signals[-1]['status'] != 'OPEN': + return None + + # Management messages - use the extended HM_SL_HIT_RE + if CANCEL_RE.search(msg): + self.signals[-1]['status'] = 'CANCELLED' + self.signals[-1]['cancel_date'] = date + self.signals[-1]['cancel_reason'] = msg.strip() + return 'CANCEL' + + # BE-move messages: base class BE_MOVE_RE catches English phrases; + # HM adds German ones via HM_SL_HIT_RE (which doubles as BE detector). + # Check HM-specific German BE patterns first, then fall through to + # the base class pattern for shared English phrases. + HM_BE_MOVE_RE = re.compile( + r'bin\s+auf\s+break\s*even' + r'|bin\s+auf\s+be\b' + r'|bin\s+\w+\s+auf\s+be\b' + r'|auf\s+be\s+raus' + r'|alle\s+(?:\w+\s+)*auf\s+be\b' + r'|sl\s+auf\s+(?:be|break\s*even)\b' + r'|jetzt\s+auf\s+be\b', + re.IGNORECASE, + ) + if HM_BE_MOVE_RE.search(msg) or BE_MOVE_RE.search(msg): + # Record on the signal but keep it OPEN (position not closed yet) + self.signals[-1]['be_move_date'] = date + return 'BE_MOVE' + + if HM_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): + # "TP1 hit, that's +70 pips" β†’ TP_NUMBER_RE finds "TP1", TP_DONE_RE finds "hit" + self._handle_tp_report(date, msg) + elif PARTIAL_CLOSE_RE.search(msg): + self._handle_partial_close(date, msg) + return None + + def _handle_sl_hit(self, date, msg): + """Override to also recognize German BE phrases as breakeven exits.""" + sig = self.signals[-1] + sig['status'] = 'CLOSED' + # Reuse HM_SL_HIT_RE but distinguish BE vs SL close type. + # Any message that reaches here already matched HM_SL_HIT_RE. + # Classify as BE if the message mentions BE/break-even (not a plain SL). + be_hit = bool(re.search( + r'\bbe\s*hit\b' + r'|bin\s+auf\s+break\s*even' + r'|bin\s+auf\s+be\b' + r'|bin\s+\w+\s+auf\s+be\b' + r'|auf\s+be\s+raus' + r'|alle\s+(?:\w+\s+)*auf\s+be\b' + r'|sl\s+auf\s+(?:be|break\s*even)\b' + r'|jetzt\s+auf\s+be\b', + msg, re.IGNORECASE, + )) + sig['reported_close_type'] = 'BE' if be_hit else 'SL' + sig['reported_close_date'] = date + + +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 the same TELEGRAM_API_ID / TELEGRAM_API_HASH / PHONE_NUMBER env + vars as live_signal_trader.py. Uses a temporary session file so it never + conflicts with the live trader's session. + """ + import os + from dotenv import load_dotenv + load_dotenv() + from datetime import 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. + # Telegram stores channel IDs as positive ints internally; the -100 + # prefix is only a convention in client-facing representations. + 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 HM channel export. + + File mode (default): + python3 hm_signal_parser.py [mt5_url] [tier] [compound] [capital] [sizing] + + Telegram mode (live history fetch – no file needed): + python3 hm_signal_parser.py --telegram [channel_id] [days] [mt5_url] [tier] [compound] [capital] [sizing] + + In Telegram mode the script connects using the same TELEGRAM_API_ID / + TELEGRAM_API_HASH / PHONE_NUMBER env vars as the live trader, fetches the + last N days of history from the HM channel, feeds every message through the + parser, then runs the full backtest analysis β€” identical output to file mode. + + channel_id defaults to the CHANNEL2_ID env var (the HM channel). + days defaults to 5. + """ + import sys + import os + from traderanalyser import RISK_TIERS, RISK_PCT_TIERS, MIN_LOT + + # ── Telegram mode ──────────────────────────────────────────────────────── + if len(sys.argv) > 1 and sys.argv[1] == '--telegram': + from dotenv import load_dotenv + load_dotenv() + argv = sys.argv[2:] # strip the flag + channel_id = int(argv[0]) if len(argv) > 0 else int(os.environ.get('CHANNEL2_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 CHANNEL2_ID env var).") + sys.exit(1) + + # Use a placeholder path – the file is never opened in Telegram mode. + analyzer = HMTradeSignalAnalyzer( + 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 'hm_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 = HMTradeSignalAnalyzer( + file_path, initial_capital=initial_capital, + compounding=compounding, mt5_base_url=mt5_base_url, + ) + analyzer.parse_signals() + + if not analyzer.signals: + print("No signals found.") + return + + if analyzer.skipped_signals: + print(f"\n{len(analyzer.skipped_signals)} signal(s) skipped by sanity check:") + 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']) + print(f"\n{len(analyzer.signals)} signals found ({n_informal} informal)") + + # Show risk hint distribution + hints = [s.get('risk_hint') for s in analyzer.signals if s.get('risk_hint')] + if hints: + from collections import Counter + print(f"Risk hints in signals: {dict(Counter(hints))}") + + 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' if compounding else 'FIXED' + print(f"\nRisk tier comparison [{mode_label}]:") + print("-" * 70) + print(" FIXED table:") + for tier in ('high', 'mid', 'low'): + df = analyzer.analyze_trades(risk_level=tier, sizing_mode='fixed') + evaluated = df[df['result'] != 'NO_DATA'] + final = df['capital'].iloc[-1] if not df.empty else initial_capital + ret = (final - initial_capital) / initial_capital * 100 + wr = evaluated['pnl'].gt(0).mean() * 100 if not evaluated.empty else 0.0 + print(f" {tier:>5}: final {final:,.2f} USDT return {ret:+.1f}% " + f"max dd {df['drawdown'].max():.1f}% win rate {wr:.1f}%") + + print(" RISK_PCT table:") + for tier in ('high', 'mid', 'low'): + df = analyzer.analyze_trades(risk_level=tier, sizing_mode='risk_pct') + evaluated = df[df['result'] != 'NO_DATA'] + final = df['capital'].iloc[-1] if not df.empty else initial_capital + ret = (final - initial_capital) / initial_capital * 100 + wr = evaluated['pnl'].gt(0).mean() * 100 if not evaluated.empty else 0.0 + print(f" {tier:>5} ({RISK_PCT_TIERS[tier]*100:.1f}%/trade): final {final:,.2f} USDT " + f"return {ret:+.1f}% max dd {df['drawdown'].max():.1f}% win rate {wr:.1f}%") + + print(" SURVIVAL table (1/3 risk, TP1-only fallback when lot < 0.01):") + 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 initial_capital + ret_surv = (final_surv - initial_capital) / initial_capital * 100 + wr_surv = evaluated_surv['pnl'].gt(0).mean() * 100 if not evaluated_surv.empty else 0.0 + n_tp1 = (df_surv['sizing'].str.startswith('survival/TP1')).sum() if 'sizing' in df_surv.columns else '?' + print(f" (compounding): final {final_surv:,.2f} USDT return {ret_surv:+.1f}% " + f"max dd {df_surv['drawdown'].max():.1f}% win rate {wr_surv:.1f}% " + f"({n_tp1} trades in TP1-only fallback)") + + 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.") + return + + print(f"\nDetailed results β€” '{chosen_tier}' tier, '{sizing_mode}' sizing:") + print("-" * 50) + final = results_df['capital'].iloc[-1] + print(f"Initial capital: {initial_capital} USDT") + print(f"Final capital: {final:.2f} USDT") + print(f"Total return: {(final - initial_capital) / initial_capital * 100:.2f}%") + print(f"Max 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: + print(f"Win rate: {evaluated['pnl'].gt(0).mean()*100:.2f}%") + print(f"Best trade: {evaluated['pnl'].max():.2f} USDT") + print(f"Worst trade: {evaluated['pnl'].min():.2f} USDT") + print(f"Avg pnl/trade: {evaluated['pnl'].mean():.2f} USDT") + + results_df.to_csv('hm_trading_results.csv', index=False) + print("\nFull per-trade results saved to hm_trading_results.csv") + + base_cols = ['date', 'pair', 'action', 'entry', 'signal_type', 'lot', + 'reported_result', 'result', 'pnl', 'pnl_pct', 'capital'] + extra = ['sizing'] if sizing_mode == 'survival' and 'sizing' in results_df.columns else [] + cols = [c for c in base_cols + extra if c in results_df.columns] + display_df = results_df[cols].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("\nDetailed trade results:") + print(display_df.to_string()) + + +if __name__ == '__main__': + main()