bast/MetaTrader5-Docker-Goldex
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
MetaTrader5-Docker-Goldex/bots/live_signal_trader.py

1216 lines
62 KiB
Python
Raw Normal View History

initial import
2026-11-06 13:22:28 +00:00
"""
Live Telegram signal trader.
Listens to one or two gold-signal Telegram channels in real time and
processes each message with the EXACT same parsing logic as
traderanalyser.py: fully detailed signals are taken as given, informal
"Gold buy now" messages get the same default SL / TP ladder, and the
same precursor-merge logic prevents a "now" announcement and its
detailed follow-up from being treated as two trades.
Two modes:
--dry (DEFAULT - safe): logs what the bot WOULD do for every signal,
and tracks a paper position against live market data so you
can watch the simulated equity curve evolve. No order is ever
sent in this mode.
--live : places REAL orders, using the exact same MCP REST calls your
live_xgboost.py already uses successfully on both BITMT501
(MT5/Vantage) and bitrtc01 (cTrader/Pepperstone) - same
/order/open, /order/modify and /positions endpoints, same
request/response shape. Since the channel's signals carry a
4-level TP ladder but a single MT5/cTrader position only
has one SL/TP pair, each signal opens one position PER TP
level (same lot/n, same SL, each with its own TP) - exactly
how scaling-out is normally done. Once the first tranche
closes, the remaining tranches' SL is moved to breakeven via
/order/modify, same as your _move_sl_to_be.
Position sizing defaults to the risk-percentage model (risk_based_lot),
not the fixed lot-per-$1000 table - see RISK_PCT_TIERS in
trade_signal_analyzer.py for why: the fixed table ignores how wide a
given trade's stop is, which is what let one wide-stop signal wipe your
backtest account in a single trade. Pass --sizing fixed to match the
backtester's original table instead, if you want a direct comparison.
Dual-channel support:
Pass --channel2-id to also listen on a second Telegram channel that
uses the HM XAUUSD format (different message syntax, German BE phrases,
REENTRY/RUNNING skipping). The second channel shares the same MT5
server, sizing config and survival rule as channel 1 - the two traders
just maintain separate open-position books.
Setup:
pip install telethon python-dotenv
Create a .env file next to this script with:
API_ID=...
API_HASH=...
PHONE_NUMBER=...
CHANNEL_ID=-1001234567890 # Goldex (channel 1)
CHANNEL2_ID=-1009876543210 # HM XAUUSD (channel 2, optional)
Usage:
python3 live_signal_trader.py # dry-run, mid tier, risk_pct sizing
python3 live_signal_trader.py --risk-tier low
python3 live_signal_trader.py --sizing fixed --risk-tier mid
python3 live_signal_trader.py --live --mt5-base-url http://localhost:8000 # REAL ORDERS
python3 live_signal_trader.py --channel2-id -1009876543210 # + HM channel
"""
import argparse
import json
import logging
import os
import sys
from collections import deque
from datetime import datetime, timezone, timedelta
import requests
from dotenv import load_dotenv
from telethon import TelegramClient, events
from traderanalyser import (
TradeSignalAnalyzer, RISK_TIERS, RISK_PCT_TIERS, CONTRACT_SIZE,
DEFAULT_CONTRACT_SIZE, PIP_SIZE, DEFAULT_PIP_SIZE, risk_based_lot,
)
try:
from hm_signal_parser import HMTradeSignalAnalyzer
except ImportError: # nur für --channel2-id nötig
HMTradeSignalAnalyzer = None
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger('live_signal_trader')
def _signal_to_dict(signal):
"""Return a JSON-serialisable copy of a parsed signal dict."""
if signal is None:
return None
d = {}
for k, v in signal.items():
if hasattr(v, 'isoformat'):
d[k] = v.isoformat()
elif isinstance(v, set):
# reported_tp_hits is a set of int indices - render as sorted list
d[k] = sorted(v)
else:
d[k] = v
return d
load_dotenv()
API_ID = os.getenv('API_ID')
API_HASH = os.getenv('API_HASH')
PHONE_NUMBER = os.getenv('PHONE_NUMBER')
CHANNEL_ID_ENV = os.getenv('CHANNEL_ID')
CHANNEL2_ID_ENV = os.getenv('CHANNEL2_ID')
class LiveSignalTrader:
"""Wraps a TradeSignalAnalyzer (used purely for its parsing/default-
fill logic and MT5 fetch helper - parse_signals()/analyze_trades()
from the backtester are never called here) with live position
tracking against fresh market data.
Survival-sizing rule
--------------------
Before opening every trade the bot fetches the REAL account balance
from /account and asks: "could I survive 3 consecutive full stop-outs
at this trade's stop distance?"
survival_lot = balance / (3 * stop_distance * contract_size)
If survival_lot < min_lot (0.01 by default): the account is too small
to safely run the full TP ladder - the bot opens a SINGLE position of
exactly 0.01 lot and exits at TP1 only, so one bad string of three
signals won't wipe the account even at minimum size.
If survival_lot >= min_lot: normal sizing applies (configured
sizing_mode / risk_tier), capped at survival_lot so a single stop-out
can never cost more than 1/3 of the current balance.
"""
def __init__(self, dry_run=True, sizing_mode='risk_pct', risk_tier='mid',
initial_capital=100, mt5_base_url='http://localhost:8000',
informal_grace_seconds=90, min_lot=0.01,
require_sl_tp=False,
analyzer_class=None, channel_name='ch1',
max_concurrent=1, diagnostics_path=None):
self.dry_run = dry_run
self.sizing_mode = sizing_mode
self.risk_tier = risk_tier
self.capital = initial_capital
self.informal_grace_seconds = informal_grace_seconds
self.min_lot = min_lot
self.require_sl_tp = require_sl_tp # when True, skip informal signals entirely
self.channel_name = channel_name # used in log prefixes for dual-channel setups
self.max_concurrent = max_concurrent # max total open positions across all siblings
self._siblings = [] # other LiveSignalTrader instances sharing the same account
# --- Diagnostics ring-buffer (last 24 h of messages + trade decisions) ---
# Each entry: {'ts': ISO str, 'raw_text': str, 'parsed': signal|None,
# 'decision': 'trade'|'skipped(concurrent)'|'informal_pending'|
# 'superseded'|'no_signal'|'resolve_failed', 'detail': str}
self._diag: deque = deque() # unbounded; pruned to 24 h on each write
self.diagnostics_path = diagnostics_path # file to write; None = don't write
analyzer_class = analyzer_class or TradeSignalAnalyzer
self.analyzer = analyzer_class(file_path=None, initial_capital=initial_capital,
mt5_base_url=mt5_base_url)
self.open_positions = [] # paper (or real) positions currently being tracked
# An informal "Gold buy now" is often just the precursor to a
# detailed follow-up seconds later (exactly like your Jan 22
# sample). In file replay we can look at the whole conversation
# at once and drop the precursor after the fact; live, we have
# to decide immediately and can't undo an order already sent -
# so an informal signal is held here for informal_grace_seconds
# before actually being opened, and a detailed signal for the
# same pair+direction arriving in that window supersedes it.
self._pending_informal = {} # (pair, action) -> signal dict, while still within its grace period
self._superseded = set() # id() of informal signals that got superseded before their grace period elapsed
# ------------------------------------------------------------------
# Diagnostics helpers
# ------------------------------------------------------------------
def _diag_record(self, ts, raw_text, parsed_signal, decision, detail=''):
"""Append one message+decision entry to the ring buffer."""
entry = {
'ts': ts.isoformat() if hasattr(ts, 'isoformat') else str(ts),
'channel': self.channel_name,
'raw_text': raw_text,
'parsed': _signal_to_dict(parsed_signal) if parsed_signal else None,
'decision': decision,
'detail': detail,
}
self._diag.append(entry)
def dump_diagnostics(self, path=None):
"""Write the last 24 h of messages and trade decisions to a JSON
file. Call this any time (e.g. via --dump-diagnostics or from the
periodic check loop). Returns the path written."""
out_path = path or self.diagnostics_path
if out_path is None:
out_path = f'diagnostics_{self.channel_name}.json'
cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
recent = [e for e in self._diag
if datetime.fromisoformat(e['ts'].replace('Z', '+00:00')) >= cutoff]
payload = {
'channel': self.channel_name,
'generated_at': datetime.now(timezone.utc).isoformat(),
'window_hours': 24,
'total_messages': len(recent),
'entries': recent,
}
with open(out_path, 'w', encoding='utf-8') as f:
json.dump(payload, f, indent=2, default=str)
logger.info(f"[{self.channel_name}] Diagnostics written → {out_path} ({len(recent)} entries)")
return out_path
def register_sibling(self, other):
"""Link two traders that share the same MT5 account so the
concurrent-position guard counts open trades from both channels."""
if other not in self._siblings:
self._siblings.append(other)
if self not in other._siblings:
other._siblings.append(self)
def _total_open(self):
"""Total open positions across this trader and all siblings.
Positions that have already moved to BE (be_done=True) are at zero
risk and do NOT block a new concurrent entry."""
at_risk = sum(1 for p in self.open_positions if not p.get('be_done', False))
at_risk += sum(
sum(1 for p in s.open_positions if not p.get('be_done', False))
for s in self._siblings
)
return at_risk
# ------------------------------------------------------------------
# Incoming messages
# ------------------------------------------------------------------
def handle_message(self, date, text):
"""Returns the parsed signal (if any) so the caller can schedule
the grace-period finalization for informal ones - see
finalize_informal() and the event handler in main()."""
logger.info(f"MSG [{date}]: {text!r}")
try:
signal = self.analyzer.process_message(date, text)
except Exception:
logger.exception("Failed to parse message - ignoring it")
self._diag_record(date, text, None, 'parse_error')
return None
if signal is None:
# Management / update messages (TP done, SL hit, partial close)
self._diag_record(date, text, None, 'no_signal',
'management/update message - no new position')
return None
if signal == 'CANCEL':
# Admin retracted the last signal - close any open position for it
self._diag_record(date, text, None, 'cancel',
'admin retracted last signal - closing position if open')
logger.warning(f"[{self.channel_name}] CANCEL message received: {text!r}")
self._handle_cancel(date)
return None
if signal == 'BE_MOVE':
# Trader said "moved all to breakeven" - relay SL → entry to MT5
self._diag_record(date, text, None, 'be_move',
'trader moved SL to breakeven - adjusting open tickets')
logger.info(f"[{self.channel_name}] BE_MOVE: {text!r}")
self._handle_be_move()
return None
key = (signal['pair'], signal['action'])
if not signal['is_informal']:
pending = self._pending_informal.pop(key, None)
if pending is not None:
self._superseded.add(id(pending))
logger.info(f"Detailed signal for {key} supersedes its informal precursor - "
f"only the detailed one will be opened")
self._diag_record(date, text, pending, 'superseded',
f'informal {key} superseded by this detailed signal')
self._open_signal(signal)
return signal
# Check if we already have an open informal position for this pair+direction.
# If so, upgrade it with the real SL/TP rather than skipping as concurrent.
informal_pos = next(
(p for p in self.open_positions
if p['signal'].get('pair') == signal['pair']
and p['signal'].get('action') == signal['action']
and p['signal'].get('is_informal')),
None
)
if informal_pos is not None:
self._upgrade_informal_position(informal_pos, signal, date, text)
return signal
self._open_signal(signal)
else:
if self.require_sl_tp:
# Strict mode: only trade signals that carry a real SL and TP1.
# Informal signals ("Gold sell now") are silently dropped here.
self._diag_record(date, text, signal, 'skipped(informal)',
'require_sl_tp=True: informal signal ignored (no SL/TP)')
logger.info(f"[{self.channel_name}] Skipping informal {signal['pair']} "
f"{signal['action']} - require_sl_tp is active")
return None
self._pending_informal[key] = signal
self._diag_record(date, text, signal, 'informal_pending',
f'waiting {self.informal_grace_seconds}s for detailed follow-up')
return signal
def _handle_be_move(self):
"""Trader said 'moved all to breakeven'. Move SL to entry on every
open ticket for this channel so our MT5 orders mirror the trader's
manual adjustment."""
if not self.open_positions:
logger.info(f"[{self.channel_name}] BE_MOVE - no open positions to adjust")
return
for pos in self.open_positions:
# Use actual fill price if available; fall back to signal entry midpoint.
# signal['entry'] is the ZONE midpoint (e.g. 3903.75 for zone 3902–3905.5),
# but the real fill may be well above/below that — using midpoint as BE
# puts SL on the wrong side of the fill and guarantees a loss.
entry = pos.get('fill_price') or pos['signal'].get('entry')
if entry is None:
logger.warning(f"[{self.channel_name}] BE_MOVE - no entry price on position, skipping")
continue
logger.info(f"[{self.channel_name}] BE_MOVE - using BE price {entry:.2f} "
f"(fill_price={pos.get('fill_price')}, signal_entry={pos['signal'].get('entry')})")
if self.dry_run:
logger.info(f"[{self.channel_name}] BE_MOVE [DRY] - would move SL → {entry} "
f"for tickets {pos.get('tickets', [])}")
pos['current_sl'] = entry
else:
for ticket in pos.get('tickets', []):
self._move_sl_to_be(ticket, entry)
pos['current_sl'] = entry
def _upgrade_informal_position(self, pos, detailed_signal, date, raw_text):
"""A detailed signal arrived after we already opened on the informal precursor.
Update the open position's SL and TP to match the real signal, and relay the
changes to MT5 via /order/modify so the live orders are corrected.
The position lot and number of tranches are NOT changed — we can't add new
orders after the fact, so we just correct the levels on the existing ticket(s).
"""
old_sl = pos['signal'].get('sl')
old_tps = list(pos['signal'].get('tp_levels', []))
new_sl = detailed_signal.get('sl')
new_tps = detailed_signal.get('tp_levels', [])
# Build a summary of what changed for logging / diagnostics
changes = []
if new_sl and old_sl and abs(new_sl - old_sl) > 0.01:
changes.append(f"SL {old_sl:.2f}→{new_sl:.2f}")
if new_tps and new_tps != old_tps:
changes.append(f"TPs {[round(t,2) for t in old_tps]}→{[round(t,2) for t in new_tps]}")
change_str = ' '.join(changes) if changes else 'no price change'
logger.info(f"[{self.channel_name}] UPGRADE informal→detailed: {change_str}")
# Update the position record
pos['signal']['sl'] = new_sl if new_sl else old_sl
if new_tps:
pos['signal']['tp_levels'] = new_tps
if new_sl:
pos['current_sl'] = new_sl
pos['signal']['is_informal'] = False # mark as upgraded
pos['upgraded_from_informal'] = True
if self.dry_run:
logger.info(f"[{self.channel_name}] UPGRADE [DRY] – position updated in memory; "
f"tickets would be modified on live account")
else:
tickets = pos.get('tickets', [])
for i, ticket in enumerate(tickets):
# Assign the TP for this tranche (index i into the new TP list)
tp = new_tps[i] if new_tps and i < len(new_tps) else None
payload = {"ticket": int(ticket)}
if new_sl:
payload["sl"] = round(float(new_sl), 2)
if tp:
payload["tp"] = round(float(tp), 2)
try:
resp = requests.post(
f"{self.analyzer.mt5_base_url}/order/modify",
json=payload, timeout=10,
)
result = resp.json()
if result.get('success'):
logger.info(f"[LIVE] UPGRADE ticket=#{ticket} {change_str}")
else:
logger.warning(f"[LIVE] UPGRADE FAILED ticket=#{ticket}: {result.get('error')}")
except Exception as e:
logger.warning(f"[LIVE] UPGRADE error ticket=#{ticket}: {e}")
self._diag_record(
date, raw_text, detailed_signal, 'upgraded',
f'informal position upgraded: {change_str} tickets={pos.get("tickets", [])} '
f'{"DRY" if self.dry_run else "LIVE"}'
)
def _handle_cancel(self, date):
"""Admin retracted the last signal. Discard any pending informal for
the same channel, and close any open position that was opened from
the most recent signal (dry: drop it; live: send /order/close)."""
# Clear any pending informal (it was the signal being cancelled)
self._pending_informal.clear()
if not self.open_positions:
logger.info(f"[{self.channel_name}] CANCEL - no open positions to close")
return
# Close the most recently opened position
pos = self.open_positions[-1]
pair = pos['signal']['pair']
action = pos['signal']['action']
if self.dry_run:
self.open_positions.pop()
logger.warning(
f"[{self.channel_name}] CANCEL [DRY] - dropped position "
f"{pair} {action} opened at {pos['signal']['entry']}"
)
else:
tickets = pos.get('tickets', [])
for ticket in tickets:
try:
resp = requests.post(
f"{self.analyzer.mt5_base_url}/order/close",
json={"ticket": int(ticket)}, timeout=10,
)
result = resp.json()
if result.get('success'):
logger.warning(f"[{self.channel_name}] CANCEL [LIVE] closed ticket=#{ticket}")
else:
logger.error(
f"[{self.channel_name}] CANCEL [LIVE] close FAILED "
f"ticket=#{ticket}: {result.get('error')}"
)
except Exception as e:
logger.error(f"[{self.channel_name}] CANCEL [LIVE] close error ticket=#{ticket}: {e}")
self.open_positions.pop()
def finalize_informal(self, signal):
"""Called by the caller's scheduled task after informal_grace_seconds.
If a detailed signal for the same pair+direction superseded this
one in the meantime, it's discarded; otherwise it's opened now."""
if id(signal) in self._superseded:
self._superseded.discard(id(signal))
# already recorded as 'superseded' in handle_message
return
key = (signal['pair'], signal['action'])
if self._pending_informal.get(key) is signal:
del self._pending_informal[key]
self._open_signal(signal)
def _fetch_account_balance(self):
"""Fetch the live account balance from the MT5/cTrader MCP server's
/account endpoint. Returns the balance as a float, or None on any
error (the caller then keeps the last known self.capital value).
The endpoint shape matches the MCP server used by live_xgboost.py:
GET /account -> {"balance": 1234.56, "equity": 1230.00, ...}
"""
try:
resp = requests.get(f"{self.analyzer.mt5_base_url}/account", timeout=10)
resp.raise_for_status()
data = resp.json()
# Accept both lowercase and title-case keys just in case
balance = data.get('balance') or data.get('Balance')
if balance is not None and float(balance) > 0:
return float(balance)
logger.warning(f"/account returned an unexpected payload: {data} - using last known balance")
except Exception as e:
logger.warning(f"Could not fetch /account: {e} - using last known balance {self.capital:.2f}")
return None
def _open_signal(self, signal):
if not self._resolve_missing_fields(signal):
logger.warning(f"Could not resolve a price for {signal['pair']} {signal['action']} "
f"at {signal['date']} - skipping, no MT5 data available")
self._diag_record(signal['date'], '', signal, 'resolve_failed',
'no live price from MT5 - signal skipped')
return
# --- Cross-channel concurrent-position guard -------------------------
# When two channels share one account the survival rule's "3 stop-outs"
# assumption breaks down if both channels are in a trade at the same
# time. Refuse to open a new position once total_open >= max_concurrent.
total_open = self._total_open()
if self.max_concurrent > 0 and total_open >= self.max_concurrent:
sibling_info = (
' (siblings: ' +
', '.join(f"{s.channel_name}={len(s.open_positions)}" for s in self._siblings) +
')'
) if self._siblings else ''
logger.warning(
f"[{self.channel_name}] SKIP {signal['pair']} {signal['action']}: "
f"total_open={total_open} >= max_concurrent={self.max_concurrent}"
f"{sibling_info} — signal ignored to protect account"
)
self._diag_record(signal['date'], '', signal, 'skipped(concurrent)',
f'total_open={total_open} >= max_concurrent={self.max_concurrent}')
return
# --- Refresh balance from the live account before every trade ---
live_balance = self._fetch_account_balance()
if live_balance is not None:
if abs(live_balance - self.capital) > 0.01:
logger.info(f"Balance updated from API: {self.capital:.2f} -> {live_balance:.2f}")
self.capital = live_balance
# --- 3-signal survival rule ------------------------------------------
# survival_lot is the largest lot where THREE consecutive full stop-outs
# would exhaust at most the entire current balance:
# 3 * survival_lot * contract_size * stop_distance = balance
# => survival_lot = balance / (3 * stop_distance * contract_size)
# which is exactly risk_based_lot(balance, 1/3, ...)
# Snapshot the parsed TP list BEFORE survival mode may truncate it,
# so the diagnostics always show what the parser actually extracted.
_parsed_tp_levels = list(signal['tp_levels'])
survival_lot = risk_based_lot(self.capital, 1.0 / 3.0,
signal['entry'], signal['sl'], signal['pair'])
if survival_lot < self.min_lot:
# WARMUP MODE: balance is too small for risk-based sizing of the
# full TP ladder, but we calculate how many min-lot tranches the
# 3-survival rule can still support at the absolute minimum lot:
#
# warmup_tranches = floor(survival_lot / min_lot)
# = floor(balance / (3 * stop_dist * contract_size * min_lot))
#
# This means:
# $30 balance → survival_lot ≈ 0.0095 → 0 → capped to 1 TP
# $50 balance → survival_lot ≈ 0.016 → 1 TP
# $70 balance → survival_lot ≈ 0.022 → 2 TPs
# $110 balance → survival_lot ≈ 0.035 → 3 TPs
# $140 balance → survival_lot ≈ 0.044 → 4 TPs
# (exact thresholds depend on stop distance of each trade)
#
# The system gradually scales up from TP1 toward the full ladder
# as capital grows, instead of staying locked at TP1-only until
# suddenly the account is large enough for full risk-based sizing.
warmup_tranches = max(1, int(survival_lot / self.min_lot))
n_tranches = warmup_tranches
actual_tps = min(warmup_tranches, len(signal['tp_levels']))
# lot = total across all tranches so _place_order splits evenly:
# round(actual_tps * min_lot / actual_tps, 2) = min_lot per tranche.
# Setting lot = min_lot caused tranche_lot = round(0.01/3,2) = 0.00
# and _place_order returned [] (no orders placed).
lot = actual_tps * self.min_lot
if actual_tps < len(signal['tp_levels']):
logger.warning(
f"{signal['pair']} {signal['action']}: balance={self.capital:.2f}, "
f"survival lot={survival_lot:.4f} < min={self.min_lot} "
f"→ WARMUP MODE: {self.min_lot} lot × {actual_tps} TP(s) "
f"(skipping TP{actual_tps+1}..TP{len(signal['tp_levels'])})"
)
else:
logger.info(
f"{signal['pair']} {signal['action']}: balance={self.capital:.2f}, "
f"survival lot={survival_lot:.4f} < min={self.min_lot} "
f"→ WARMUP MODE: {self.min_lot} lot × {actual_tps} TP(s) (all parsed TPs fit)"
)
signal['tp_levels'] = signal['tp_levels'][:actual_tps]
else:
# Normal sizing, capped at the survival threshold so a single
# trade can never cost more than 1/3 of the current balance.
configured_lot = self._lot_for(signal)
if configured_lot > survival_lot:
logger.info(
f"{signal['pair']} {signal['action']}: configured lot {configured_lot:.4f} "
f"exceeds survival cap {survival_lot:.4f} (1/3 of {self.capital:.2f}) "
f"→ capping to survival lot"
)
raw_lot = min(configured_lot, survival_lot)
if raw_lot < self.min_lot:
# Configured risk_pct is so conservative that the resulting lot
# is below min_lot even though survival_lot allows more.
# Use the same warmup ladder logic: open as many min-lot tranches
# as the survival budget supports, so tight-SL signals with a
# healthy balance still run the full (or partial) TP ladder
# instead of being forced to a single TP by the round-up.
# Use survival_lot as total position size (Kelly-optimal):
# each tranche = floor(survival_lot / n / min_lot) × min_lot
# so we use the full 1/3-of-balance safety budget, not just
# n × min_lot. This gives geometric growth as balance rises:
# $101→$200 same pace, $200→$300 2× faster, $300→ 3×, etc.
warmup_tranches = max(1, int(survival_lot / self.min_lot))
actual_tps = min(warmup_tranches, len(signal['tp_levels']))
n_tranches = actual_tps
# tranche_lot = floor to valid lot step (0.01), total ≤ survival_lot
tranche_lot_raw = survival_lot / actual_tps
tranche_lot = int(tranche_lot_raw / self.min_lot) * self.min_lot
tranche_lot = max(tranche_lot, self.min_lot)
lot = actual_tps * tranche_lot
if actual_tps < len(signal['tp_levels']):
logger.warning(
f"{signal['pair']} {signal['action']}: configured lot {raw_lot:.4f} < min={self.min_lot}, "
f"survival budget allows {warmup_tranches} tranche(s) "
f"→ WARMUP(survival): {tranche_lot:.2f} lot × {actual_tps} TP(s) "
f"(survival_lot={survival_lot:.4f})"
)
else:
logger.info(
f"{signal['pair']} {signal['action']}: configured lot {raw_lot:.4f} < min={self.min_lot} "
f"→ WARMUP(survival): {tranche_lot:.2f} lot × {actual_tps} TP(s) "
f"(survival_lot={survival_lot:.4f})"
)
signal['tp_levels'] = signal['tp_levels'][:actual_tps]
else:
lot, n_tranches = self._resolve_tradable_size(raw_lot, len(signal['tp_levels']))
if lot <= 0:
logger.warning(f"Computed lot ({raw_lot:.4f}) is below the minimum tradable lot "
f"({self.min_lot}) even as a single position - skipping "
f"{signal['pair']} {signal['action']}")
return
if n_tranches < len(signal['tp_levels']):
step = len(signal['tp_levels']) / n_tranches
signal['tp_levels'] = [signal['tp_levels'][min(int(i * step), len(signal['tp_levels']) - 1)]
for i in range(n_tranches)]
logger.warning(f"{signal['pair']} {signal['action']}: lot {raw_lot:.4f} too small to split "
f"across the full TP ladder at min lot {self.min_lot} - consolidated to "
f"{n_tranches} tranche(s): {[round(t, 2) for t in signal['tp_levels']]}")
tag = 'DRY' if self.dry_run else 'LIVE'
_wt = max(1, int(survival_lot / self.min_lot))
_configured_lot_for_mode = self._lot_for(signal) if survival_lot >= self.min_lot else 0
if survival_lot < self.min_lot or _configured_lot_for_mode < self.min_lot:
mode = f'warmup({len(signal["tp_levels"])}tp/0.01)'
else:
mode = f'{self.sizing_mode}/{self.risk_tier}'
logger.info(f"[{tag}] OPEN {signal['action']} {signal['pair']} @ {signal['entry']:.2f} "
f"SL {signal['sl']:.2f} TP {[round(t, 2) for t in signal['tp_levels']]} "
f"lot={lot:.4f} balance={self.capital:.2f} "
f"sizing={mode} ({'informal/defaulted' if signal['is_informal'] else 'detailed'})")
import copy as _copy
signal_snapshot = _copy.copy(signal)
signal_snapshot['tp_levels'] = _parsed_tp_levels # all TPs as parsed, before survival truncation
signal_snapshot['tp_levels_used'] = list(signal['tp_levels']) # TPs actually traded
self._diag_record(signal['date'], '', signal_snapshot, 'trade',
f"lot={lot:.4f} mode={mode} balance={self.capital:.2f} "
f"parsed_tps={len(_parsed_tp_levels)} used_tps={len(signal['tp_levels'])} "
f"{'DRY' if self.dry_run else 'LIVE'}")
if self.dry_run:
position = {
'signal': signal,
'lot': lot,
'remaining': list(range(len(signal['tp_levels']))),
'current_sl': signal['sl'],
'last_checked': signal['date'].replace(tzinfo=None),
'realized_pnl': 0.0,
}
self.open_positions.append(position)
else:
tickets = self._place_order(signal, lot)
if not tickets:
logger.error(f"No tranches were actually opened for {signal['pair']} {signal['action']} - "
f"not tracking this signal")
return
# Fetch actual fill prices so BE uses real fill, not signal entry midpoint
fill_price = self._fetch_avg_fill(tickets, signal['entry'])
logger.info(f"[LIVE] avg fill price for BE: {fill_price:.2f} "
f"(signal entry midpoint was {signal['entry']:.2f})")
self.open_positions.append({
'signal': signal,
'tickets': list(tickets),
'n_original': len(tickets),
'be_done': False,
'fill_price': fill_price,
})
def _resolve_tradable_size(self, raw_lot, n_desired_tranches):
"""Enforce the broker's minimum tradable lot (default 0.01).
If the raw risk-based lot is below min_lot even as a single
position, returns (0, 0) - this signal genuinely can't be traded
at this account size and stop width. If it IS tradable as one
position but too small to split across the full TP ladder, the
ladder gets consolidated down to however many tranches DO fit.
Rounding raw_lot UP to min_lot means the actual risk on that
trade exceeds your configured risk_pct - an unavoidable
consequence of trading a small account against a wide stop, not
a bug, and it's logged clearly when it happens."""
if raw_lot <= 0:
return 0.0, 0
total_lot = max(raw_lot, self.min_lot)
if raw_lot < self.min_lot:
logger.warning(f"Lot {raw_lot:.4f} is below the minimum tradable lot ({self.min_lot}) - "
f"rounding up, so actual risk here exceeds the configured "
f"'{self.risk_tier}' risk_pct")
max_tranches = max(1, int(round(total_lot / self.min_lot, 6)))
n_tranches = min(n_desired_tranches, max_tranches)
return total_lot, n_tranches
# Maximale Abweichung eines geparsten Einstiegs vom Live-Kurs, bevor er als
# Parse-Fehler gilt (Gold bewegt sich intraday selten >2 %).
MAX_ENTRY_DEVIATION = 0.02
def _resolve_missing_fields(self, signal):
"""Same default-fill logic as the backtester's
_resolve_informal_signals, just resolved against a live price
instead of historical bars.
Zusätzlich: ein geparster Einstieg, der weit vom Live-Kurs entfernt
ist (z.B. 2077.75 statt 4127 durch "4126.5-29"), wird verworfen und
durch den Live-Kurs ersetzt. SL/TP, die dann auf der falschen Seite
oder unplausibel weit weg liegen, werden neu aus dem Default-Plan
gebildet bzw. das Signal wird übersprungen."""
live = self._get_live_price(signal['pair'], signal['action'])
if signal['entry'] is not None and live:
dev = abs(signal['entry'] - live) / live
if dev > self.MAX_ENTRY_DEVIATION:
logger.warning(f"[{self.channel_name}] Entry {signal['entry']} weicht {dev:.0%} vom "
f"Live-Kurs {live} ab – Parse-Fehler, verwende Live-Kurs")
signal['entry'] = live
signal['tp_levels'] = [t for t in signal['tp_levels']
if abs(t - live) / live <= self.MAX_ENTRY_DEVIATION]
if signal['entry'] is not None and signal['sl'] is not None:
e, sl = signal['entry'], signal['sl']
wrong_side = (signal['action'] == 'BUY' and sl >= e) or (signal['action'] == 'SELL' and sl <= e)
if wrong_side or abs(sl - e) / e > self.MAX_ENTRY_DEVIATION:
logger.warning(f"[{self.channel_name}] SL {sl} unplausibel zu Entry {e} – Signal übersprungen")
return False
if signal['entry'] is None:
price = live or self._get_live_price(signal['pair'], signal['action'])
if price is None:
return False
signal['entry'] = price
pip = PIP_SIZE.get(signal['pair'], DEFAULT_PIP_SIZE)
if signal['sl'] is None:
dist = self.analyzer.default_sl_pips * pip
signal['sl'] = signal['entry'] - dist if signal['action'] == 'BUY' else signal['entry'] + dist
signal['tp_levels'] = [t for t in signal['tp_levels']
if (signal['action'] == 'BUY' and t > signal['entry'])
or (signal['action'] == 'SELL' and t < signal['entry'])]
if not signal['tp_levels']:
levels = [signal['entry'] + (d * pip if signal['action'] == 'BUY' else -d * pip)
for d in self.analyzer.default_tp_ladder_pips]
signal['tp_levels'] = sorted(levels, reverse=(signal['action'] == 'SELL'))
return True
def _lot_for(self, signal):
if self.sizing_mode == 'risk_pct':
lot = risk_based_lot(self.capital, RISK_PCT_TIERS[self.risk_tier],
signal['entry'], signal['sl'], signal['pair'])
else:
lot = (self.capital / 1000.0) * RISK_TIERS[self.risk_tier]
return round(lot, 4)
def _get_live_price(self, pair, action=None):
"""Live quote via the confirmed /tick/{symbol} endpoint (same one
your MT5/cTrader MCP servers expose). Uses ask for BUY, bid for
SELL - falls back to the latest M1 bar's close if the tick
endpoint is unavailable for some reason."""
try:
resp = requests.get(f"{self.analyzer.mt5_base_url}/tick/{pair}", timeout=10)
resp.raise_for_status()
tick = resp.json()
bid, ask = tick.get('bid'), tick.get('ask')
if bid and ask:
return float(ask) if action == 'BUY' else float(bid)
except Exception as e:
logger.warning(f"/tick/{pair} failed ({e}) - falling back to latest M1 close")
df = self.analyzer._mt5_fetch_ohlc(pair, 'M1', bars=1)
if df is None or df.empty:
return None
return float(df.iloc[-1]['Close'])
# ------------------------------------------------------------------
# Position monitoring
# ------------------------------------------------------------------
def check_open_positions(self):
"""Call periodically (see the polling loop in main()).
Dry-run: advances the local paper-trading simulation using fresh
bars (same fill rules as the backtester's _simulate_trade).
Live: polls the real /positions endpoint - a tranche disappearing
means it closed (TP or SL), and once the first tranche is gone
the remaining ones' SL is moved to breakeven via /order/modify,
exactly like your _move_sl_to_be."""
if self.dry_run:
self._check_open_positions_dry()
else:
self._check_open_positions_live()
def _check_open_positions_dry(self):
still_open = []
for pos in self.open_positions:
try:
closed = self._advance_position(pos)
except Exception:
logger.exception(f"Error checking position {pos['signal']['pair']} "
f"{pos['signal']['action']} - leaving it open")
closed = False
if not closed:
still_open.append(pos)
self.open_positions = still_open
def _check_open_positions_live(self):
try:
resp = requests.get(f"{self.analyzer.mt5_base_url}/positions", timeout=10)
resp.raise_for_status()
live_tickets = {p['ticket'] for p in resp.json().get('positions', [])}
except Exception as e:
logger.warning(f"Could not fetch /positions: {e} - leaving positions as tracked")
return
still_open = []
for pos in self.open_positions:
before = len(pos['tickets'])
pos['tickets'] = [t for t in pos['tickets'] if t in live_tickets]
closed_now = before - len(pos['tickets'])
if closed_now:
logger.info(f"[LIVE] {closed_now} tranche(s) closed for {pos['signal']['pair']} "
f"{pos['signal']['action']} ({len(pos['tickets'])} still open)")
if not pos['tickets']:
logger.info(f"[LIVE] All tranches closed for {pos['signal']['pair']} {pos['signal']['action']}")
continue # fully closed - drop tracking
if not pos['be_done'] and len(pos['tickets']) < pos['n_original']:
# at least one tranche closed (the first TP, in the normal
# case) - move the remaining tranches' SL to breakeven.
# Use actual fill price (stored at open), NOT signal entry
# midpoint: if fill was 3996.73 but signal entry was 3998.50,
# setting BE to 3998.50 puts SL above fill → instant loss.
be_price = pos.get('fill_price', pos['signal']['entry'])
for ticket in pos['tickets']:
self._move_sl_to_be(ticket, be_price)
pos['be_done'] = True
still_open.append(pos)
self.open_positions = still_open
def _fetch_avg_fill(self, tickets, fallback_entry):
"""Query /positions right after placement to get actual fill prices.
Returns the average open price across the given tickets, or
fallback_entry if the API call fails or no matching positions found."""
try:
resp = requests.get(f"{self.analyzer.mt5_base_url}/positions", timeout=10)
resp.raise_for_status()
positions = resp.json().get('positions', [])
ticket_set = set(int(t) for t in tickets)
fills = [float(p.get('open_price') or p.get('price_open')) for p in positions
if int(p.get('ticket', -1)) in ticket_set
and (p.get('open_price') or p.get('price_open'))]
if fills:
avg = sum(fills) / len(fills)
logger.info(f"[LIVE] fill prices from /positions: {fills} → avg {avg:.2f}")
return round(avg, 2)
except Exception as e:
logger.warning(f"[LIVE] _fetch_avg_fill failed ({e}) - using signal entry {fallback_entry:.2f}")
return float(fallback_entry)
def _move_sl_to_be(self, ticket, entry_price):
"""Move SL to breakeven via /order/modify - same call and same
payload shape as your _move_sl_to_be in live_xgboost.py."""
entry_rounded = round(float(entry_price), 2)
try:
resp = requests.post(f"{self.analyzer.mt5_base_url}/order/modify", json={
"ticket": int(ticket),
"sl": entry_rounded,
}, timeout=10)
result = resp.json()
if result.get('success'):
logger.info(f"[LIVE] BE activated ticket=#{ticket} SL -> {entry_rounded}")
else:
logger.warning(f"[LIVE] BE FAILED ticket=#{ticket}: {result.get('error')}")
except Exception as e:
logger.warning(f"[LIVE] BE error ticket=#{ticket}: {e}")
def _advance_position(self, pos):
signal = pos['signal']
df = self.analyzer._mt5_fetch_ohlc(signal['pair'], 'M1', bars=1000)
if df is None or df.empty:
return False
new_bars = df[df.index > pos['last_checked']]
if new_bars.empty:
return False
action = signal['action']
entry_price = signal['entry']
tp_levels = signal['tp_levels']
contract_size = CONTRACT_SIZE.get(signal['pair'], DEFAULT_CONTRACT_SIZE)
tranche = (pos['lot'] * contract_size) / max(len(tp_levels), 1)
tag = 'DRY' if self.dry_run else 'LIVE'
for ts, bar in new_bars.iterrows():
high, low, close = float(bar['High']), float(bar['Low']), float(bar['Close'])
sl_triggered = (action == 'BUY' and close <= pos['current_sl']) or \
(action == 'SELL' and close >= pos['current_sl'])
if sl_triggered and pos['remaining']:
for _ in pos['remaining']:
pos['realized_pnl'] += tranche * ((pos['current_sl'] - entry_price) if action == 'BUY'
else (entry_price - pos['current_sl']))
label = 'BE' if pos['current_sl'] == entry_price else 'SL'
pos['remaining'] = []
self._close_position(pos, label)
return True
for idx in list(pos['remaining']):
tp_price = tp_levels[idx]
hit = (action == 'BUY' and high >= tp_price) or (action == 'SELL' and low <= tp_price)
if hit:
pos['realized_pnl'] += tranche * ((tp_price - entry_price) if action == 'BUY'
else (entry_price - tp_price))
pos['remaining'].remove(idx)
logger.info(f"[{tag}] TP{idx + 1} hit on {signal['pair']} {action} @ {tp_price:.2f} "
f"running pnl={pos['realized_pnl']:.2f}")
if pos['current_sl'] != entry_price:
pos['current_sl'] = entry_price # breakeven after first confirmed TP
pos['last_checked'] = ts
if not pos['remaining']:
self._close_position(pos, 'ALL_TP')
return True
return False
def _close_position(self, pos, label):
tag = 'DRY' if self.dry_run else 'LIVE'
self.capital = max(0.0, self.capital + pos['realized_pnl'])
logger.info(f"[{tag}] CLOSE {pos['signal']['pair']} {pos['signal']['action']} ({label}) "
f"pnl={pos['realized_pnl']:.2f} capital={self.capital:.2f}")
# ------------------------------------------------------------------
# Real execution - confirmed against your live_xgboost.py's own calls
# ------------------------------------------------------------------
def _place_order(self, signal, lot):
"""Opens one real position per TP level (same lot/n each, same
SL, each with its own TP) via /order/open - matches your
_execute_order_mt5: POST {symbol, side, volume, sl, tp, comment},
response {"success": bool, "order": <ticket>, ...}. Returns the
list of tickets that were actually opened (a partial list if one
tranche's request failed - logged either way)."""
n = len(signal['tp_levels'])
tranche_lot = round(lot / n, 2)
if tranche_lot <= 0:
logger.error(f"Tranche lot rounds to 0.00 (lot={lot}, {n} tranches) - "
f"capital too low for this stop distance, skipping real order")
return []
side = 'buy' if signal['action'] == 'BUY' else 'sell'
tickets = []
for tp in signal['tp_levels']:
try:
resp = requests.post(f"{self.analyzer.mt5_base_url}/order/open", json={
"symbol": signal['pair'],
"side": side,
"volume": tranche_lot,
"sl": round(signal['sl'], 2),
"tp": round(tp, 2),
"comment": "telegram-signal",
}, timeout=10)
result = resp.json()
except Exception as e:
logger.error(f"[LIVE] order/open request failed for tp={tp}: {e}")
continue
if result.get('success'):
ticket = result.get('order')
tickets.append(ticket)
logger.info(f"[LIVE] order placed ticket=#{ticket} {side} {tranche_lot} {signal['pair']} "
f"sl={signal['sl']:.2f} tp={tp:.2f}")
else:
logger.error(f"[LIVE] order/open FAILED for tp={tp}: "
f"{result.get('error')} - {result.get('description')}")
return tickets
def main():
parser = argparse.ArgumentParser(description="Live Telegram signal trader")
parser.add_argument('--dry', action='store_true',
help='Explicit dry-run flag (this is also the default if neither flag is given)')
parser.add_argument('--live', action='store_true',
help='Disable dry-run and place REAL orders via /order/open on --mt5-base-url')
parser.add_argument('--sizing', choices=['risk_pct', 'fixed'], default='risk_pct',
help="Position sizing model (default risk_pct - see module docstring for why)")
parser.add_argument('--risk-tier', choices=['high', 'mid', 'low'], default='mid')
parser.add_argument('--initial-capital', type=float, default=100)
parser.add_argument('--mt5-base-url', default='http://localhost:8000')
parser.add_argument('--channel-id', type=int, default=int(CHANNEL_ID_ENV) if CHANNEL_ID_ENV else None,
help='Telegram channel ID for channel 1 (Goldex format). '
'Defaults to CHANNEL_ID env var.')
parser.add_argument('--channel2-id', type=int,
default=int(CHANNEL2_ID_ENV) if CHANNEL2_ID_ENV else None,
help='Optional second channel (HM XAUUSD format). '
'Defaults to CHANNEL2_ID env var. '
'When set, a separate trader instance with the HM parser listens here.')
parser.add_argument('--poll-interval', type=int, default=10,
help='Seconds between checks of open positions against fresh market data')
parser.add_argument('--informal-grace-seconds', type=int, default=90,
help='How long to hold an informal ("buy now") signal before opening it, '
'in case a detailed follow-up supersedes it. '
'Default 90s: ch1-Goldex detailed signals typically arrive 30-60s after '
'the informal "buy/sell now" precursor (observed in diagnostics).')
parser.add_argument('--min-lot', type=float, default=0.01,
help="Broker's minimum tradable lot - the TP ladder is consolidated to fewer "
"tranches (or the signal skipped) if the risk-based lot can't meet this")
parser.add_argument('--channels', choices=['ch1', 'ch2', 'both'], default='both',
help="Which channel(s) to listen on. "
"'ch1' = Goldex only, 'ch2' = HM only (requires --channel2-id), "
"'both' = both channels (default). "
"When channel2-id is not set 'both' and 'ch2' fall back to ch1.")
parser.add_argument('--max-concurrent', type=int, default=1,
help="Maximum total open positions allowed across ALL active channels at once. "
"A new signal is skipped (with a warning) when this limit is already reached. "
"Default 1 = never open a second position while one is already open. "
"Set to 0 to disable the guard entirely.")
parser.add_argument('--diagnostics-dir', default='.',
help="Directory where diagnostics_<channel>.json files are written. "
"Default: current directory. Files are updated every --diag-interval seconds "
"and also on clean shutdown (Ctrl-C).")
parser.add_argument('--diag-interval', type=int, default=300,
help="Seconds between automatic diagnostics file writes (default 300 = every 5 min). "
"Set to 0 to disable periodic writes (file is still written on shutdown).")
parser.add_argument('--require-sl-tp', action='store_true', default=False,
help="Skip informal signals ('Gold sell now') entirely - only open trades when "
"a full signal with SL and at least TP1 is present. "
"Eliminates auto-generated SL/TP guesses but may miss fast-moving entries.")
args = parser.parse_args()
if args.channel_id is None and args.channels in ('ch1', 'both') and not args.channel2_id:
logger.error("No channel ID given - pass --channel-id or set CHANNEL_ID in .env")
sys.exit(1)
if not (API_ID and API_HASH and PHONE_NUMBER):
logger.error("API_ID / API_HASH / PHONE_NUMBER must be set in .env (same as your existing Telethon script)")
sys.exit(1)
dry_run = not args.live
if not dry_run:
print(f"LIVE mode will place REAL orders on {args.mt5_base_url} - one position per TP level, "
f"real money at risk.")
import sys as _sys
if _sys.stdin.isatty():
confirm = input("Type 'yes' to continue: ")
if confirm.strip().lower() != 'yes':
print("Aborted.")
return
else:
print("Non-interactive mode — live trading auto-confirmed.")
logger.info("LIVE mode auto-confirmed (non-interactive / nohup)")
# Resolve which channels to actually activate
run_ch1 = args.channels in ('ch1', 'both') and args.channel_id is not None
run_ch2 = args.channels in ('ch2', 'both') and bool(args.channel2_id)
if run_ch2 and HMTradeSignalAnalyzer is None:
logger.error("--channel2-id gesetzt, aber hm_signal_parser.py fehlt im Image")
sys.exit(1)
if args.channels in ('ch2', 'both') and not args.channel2_id:
logger.warning("--channels includes ch2 but no --channel2-id / CHANNEL2_ID set — ch2 disabled")
if not run_ch1 and not run_ch2:
logger.error("No active channel after applying --channels filter — nothing to listen to")
sys.exit(1)
os.makedirs(args.diagnostics_dir, exist_ok=True)
def _diag_path(channel_name):
return os.path.join(args.diagnostics_dir, f'diagnostics_{channel_name}.json')
# --- Channel 1: Goldex (TradeSignalAnalyzer) --------------------------------
trader = None
if run_ch1:
trader = LiveSignalTrader(dry_run=dry_run, sizing_mode=args.sizing, risk_tier=args.risk_tier,
initial_capital=args.initial_capital, mt5_base_url=args.mt5_base_url,
informal_grace_seconds=args.informal_grace_seconds,
min_lot=args.min_lot, require_sl_tp=args.require_sl_tp,
channel_name='ch1-Goldex',
max_concurrent=args.max_concurrent,
diagnostics_path=_diag_path('ch1-Goldex'))
# --- Channel 2: HM XAUUSD (HMTradeSignalAnalyzer) -------------------------
trader2 = None
if run_ch2:
trader2 = LiveSignalTrader(dry_run=dry_run, sizing_mode=args.sizing, risk_tier=args.risk_tier,
initial_capital=args.initial_capital, mt5_base_url=args.mt5_base_url,
informal_grace_seconds=args.informal_grace_seconds,
min_lot=args.min_lot, require_sl_tp=args.require_sl_tp,
analyzer_class=HMTradeSignalAnalyzer,
channel_name='ch2-HM', max_concurrent=args.max_concurrent,
diagnostics_path=_diag_path('ch2-HM'))
# Link siblings so each trader's concurrent guard sees the other's positions
if trader is not None and trader2 is not None:
trader.register_sibling(trader2)
# --- Startup balance fetch --------------------------------------------------
# One account → fetch once and apply to whichever traders are active.
ref_trader = trader or trader2 # at least one is not None (checked above)
startup_balance = ref_trader._fetch_account_balance()
for t in (trader, trader2):
if t is not None:
if startup_balance is not None:
t.capital = startup_balance
# else: keeps args.initial_capital as seed
if startup_balance is not None:
logger.info(f"Account balance from API: {startup_balance:.2f} USDT "
f"({'seed' if startup_balance == args.initial_capital else 'live'} value)")
else:
logger.warning(f"Could not reach {args.mt5_base_url}/account - "
f"using seed capital {args.initial_capital:.2f} USDT for sizing until the "
f"first signal fires (will retry before each trade)")
client = TelegramClient(os.getenv('TG_SESSION', 'mysession'), API_ID, API_HASH)
# --- Event handlers --------------------------------------------------------
if trader is not None:
@client.on(events.NewMessage(chats=args.channel_id))
async def handler_ch1(event):
import asyncio
try:
date = event.message.date # tz-aware UTC datetime from Telethon
text = event.message.text or ''
signal = trader.handle_message(date, text)
if signal is not None and signal['is_informal']:
async def delayed_finalize():
await asyncio.sleep(args.informal_grace_seconds)
trader.finalize_informal(signal)
client.loop.create_task(delayed_finalize())
except Exception:
logger.exception("[ch1-Goldex] Error processing incoming message")
if trader2 is not None:
@client.on(events.NewMessage(chats=args.channel2_id))
async def handler_ch2(event):
import asyncio
try:
date = event.message.date
text = event.message.text or ''
signal = trader2.handle_message(date, text)
if signal is not None and signal['is_informal']:
async def delayed_finalize2():
await asyncio.sleep(args.informal_grace_seconds)
trader2.finalize_informal(signal)
client.loop.create_task(delayed_finalize2())
except Exception:
logger.exception("[ch2-HM] Error processing incoming message")
def dump_all_diagnostics():
"""Write diagnostics files for all active traders."""
for t in (trader, trader2):
if t is not None:
try:
t.dump_diagnostics()
except Exception as e:
logger.warning(f"Could not write diagnostics for {t.channel_name}: {e}")
async def periodic_check():
import asyncio
last_diag = 0.0
while True:
await asyncio.sleep(args.poll_interval)
# Refresh balance once; both traders share the same account.
live_bal = ref_trader._fetch_account_balance()
if live_bal is not None:
for t in (trader, trader2):
if t is not None:
t.capital = live_bal
tag = 'DRY' if dry_run else 'LIVE'
parts = []
if trader is not None: parts.append(f"ch1_open={len(trader.open_positions)}")
if trader2 is not None: parts.append(f"ch2_open={len(trader2.open_positions)}")
logger.info(f"[{tag}] Balance: {live_bal:.2f} USDT " + " ".join(parts) +
f" total_open={ref_trader._total_open()}"
f" max_concurrent={args.max_concurrent}")
for t in (trader, trader2):
if t is not None:
t.check_open_positions()
# Periodic diagnostics dump
import time
now = time.monotonic()
if args.diag_interval > 0 and (now - last_diag) >= args.diag_interval:
dump_all_diagnostics()
last_diag = now
# ------------------------------------------------------------------
# Start Telethon and enter the live-listener loop
# ------------------------------------------------------------------
client.start(phone=PHONE_NUMBER)
client.loop.create_task(periodic_check())
active = []
if trader is not None: active.append(f"ch1={args.channel_id} [Goldex]")
if trader2 is not None: active.append(f"ch2={args.channel2_id} [HM XAUUSD]")
logger.info(
f"Active channels: {', '.join(active)} | "
f"{'DRY-RUN' if dry_run else 'LIVE'} "
f"(sizing={args.sizing}, risk_tier={args.risk_tier}, "
f"seed_capital={args.initial_capital}, min_lot={args.min_lot}, "
f"max_concurrent={args.max_concurrent}) "
f"Survival rule: balance fetched live before each trade; "
f"if survival_lot < {args.min_lot} → warmup mode (0.01 lot, TPs scaled to balance)"
)
logger.info(f"Diagnostics: writing to {args.diagnostics_dir}/ every "
f"{args.diag_interval}s (also on Ctrl-C shutdown)")
try:
client.run_until_disconnected()
finally:
# Always write a final diagnostics snapshot on shutdown so the
# last session is available for analysis even if no periodic write
# fired yet.
dump_all_diagnostics()
logger.info("Shutdown complete - diagnostics saved.")
if __name__ == '__main__':
main()
Reference in a new issue Copy permalink