ea2e37f0a7
All checks were successful
MT5 Helm GitOps Deploy / build-and-deploy (push) Has been skipped
446 lines
19 KiB
Python
446 lines
19 KiB
Python
"""
|
||
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()
|