MetaTrader5-Docker-Goldex/bots/hm_signal_parser.py

"""
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 <file> [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='<telegram>',
            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()