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