1059 lines
50 KiB
Python
1059 lines
50 KiB
Python
import pandas as pd
|
||
import re
|
||
import statistics
|
||
from datetime import datetime
|
||
import requests
|
||
|
||
|
||
# Map however the provider writes a pair/asset name to a canonical code.
|
||
PAIR_ALIASES = {
|
||
'GOLD': 'XAUUSD', 'XAU': 'XAUUSD', 'XAUUSD': 'XAUUSD',
|
||
'SILVER': 'XAGUSD', 'XAG': 'XAGUSD', 'XAGUSD': 'XAGUSD',
|
||
'BTC': 'BTCUSD', 'BITCOIN': 'BTCUSD', 'BTCUSD': 'BTCUSD',
|
||
}
|
||
|
||
# Canonical pair -> symbol name as known by your MT5 terminal/broker.
|
||
# Identity mapping for almost everything (your broker uses "XAUUSD" etc.
|
||
# directly - confirmed via your /ohlc/XAUUSD sweep). Only fill this in if
|
||
# a particular broker uses a suffixed symbol (e.g. "XAUUSD.a").
|
||
MT5_SYMBOL_OVERRIDES = {
|
||
# 'XAUUSD': 'XAUUSD.a',
|
||
}
|
||
|
||
# Timeframes to try, finest first, falling back to coarser ones until one
|
||
# actually reaches back far enough to cover a given signal's date. Based
|
||
# on your own /ohlc history-depth sweep on BITMT501 (M1 ~7 days, M5 ~2
|
||
# months, M15 ~5 months, H1 ~2 years, H4 ~7 years, D1 ~back to 2007).
|
||
MT5_TIMEFRAME_FALLBACK_CHAIN = ['M15', 'H1', 'H4', 'D1']
|
||
|
||
# Lot size per $1000 of account equity, taken directly from your own
|
||
# money-management table ($100->0.10/0.05/0.01, $200->0.20/0.10/0.02,
|
||
# $500->0.50/0.25/0.05, $1000->1.00/0.50/0.10 - all the same ratio).
|
||
#
|
||
# IMPORTANT CAVEAT: this sizes lot purely off capital and ignores how far
|
||
# away the stop is for any given trade. Most of your signals use a tight
|
||
# stop, but the sanity check allows entries up to PRICE_SANITY_PCT (12%)
|
||
# from entry, and a handful of real signals do have much wider stops than
|
||
# the 60-pip default. At the high tier, lot = capital/1000, so a stop
|
||
# that's $480 away (12% of a $4000 gold price) loses 480*lot*100 = 48x
|
||
# capital in that ONE trade - which is exactly how the high-tier run hit
|
||
# exactly $0.00. This table is kept for comparison; RISK_PCT_TIERS below
|
||
# is the safer alternative actually used by the live trader.
|
||
RISK_TIERS = {
|
||
'high': 1.00,
|
||
'mid': 0.50,
|
||
'low': 0.10,
|
||
}
|
||
|
||
# Safer alternative: risk a FIXED percentage of current capital on every
|
||
# trade, with lot size computed from that trade's ACTUAL stop distance
|
||
# rather than a one-size-fits-all table. This caps the worst case loss on
|
||
# any single trade at risk_pct of capital, no matter how wide that
|
||
# particular trade's stop happens to be - eliminating the single-trade
|
||
# wipeout risk that RISK_TIERS carries.
|
||
RISK_PCT_TIERS = {
|
||
'high': 0.03, # risk 3% of capital per trade
|
||
'mid': 0.015, # 1.5%
|
||
'low': 0.005, # 0.5%
|
||
}
|
||
|
||
|
||
def risk_based_lot(capital, risk_pct, entry, sl, pair):
|
||
"""lot size such that a full stop-out loses exactly risk_pct of
|
||
capital, regardless of how wide this particular trade's stop is."""
|
||
contract_size = CONTRACT_SIZE.get(pair, DEFAULT_CONTRACT_SIZE)
|
||
stop_distance = abs(entry - sl)
|
||
if stop_distance <= 0:
|
||
return 0.0
|
||
return (capital * risk_pct) / (stop_distance * contract_size)
|
||
|
||
# Contract size (units per 1.00 lot) used to convert a price move into a
|
||
# dollar P&L: pnl = lot * contract_size * price_difference. 100 oz/lot is
|
||
# the common CFD convention for XAUUSD on brokers like Vantage/Pepperstone
|
||
# - double-check yours via the MT5 symbol specification if unsure.
|
||
CONTRACT_SIZE = {
|
||
'XAUUSD': 100,
|
||
'XAGUSD': 5000,
|
||
'BTCUSD': 1,
|
||
}
|
||
DEFAULT_CONTRACT_SIZE = 100_000 # standard FX lot, units of base currency
|
||
|
||
# Price distance of one "pip" per pair, used to translate your stated
|
||
# default risk plan (60 pip SL, 20/40/60/80 pip TP ladder) into actual
|
||
# price levels for informal signals ("Gold buy now") that don't carry
|
||
# their own SL/TP. Derived from your real signals' TP spacing ($2 per 20
|
||
# pips on XAUUSD, i.e. pip=$0.1).
|
||
PIP_SIZE = {
|
||
'XAUUSD': 0.1,
|
||
'XAGUSD': 0.01,
|
||
'BTCUSD': 1.0,
|
||
}
|
||
DEFAULT_PIP_SIZE = 0.0001 # standard FX pairs
|
||
|
||
# Smallest tradable lot at most retail brokers (Vantage/Pepperstone both
|
||
# use 0.01). Used by the 'survival' sizing mode to decide when to fall
|
||
# back to the minimal-size / TP1-only path.
|
||
MIN_LOT = 0.01
|
||
|
||
# If an informal "now" signal is followed within this many seconds by a
|
||
# fully detailed signal for the same pair/direction, the informal one was
|
||
# just the precursor announcement (e.g. "Gold sell now" immediately
|
||
# followed by the full SL/TP message) - it gets dropped so the same setup
|
||
# isn't counted as two separate trades.
|
||
INFORMAL_MERGE_WINDOW_SECONDS = 15 * 60
|
||
|
||
# A freshly-parsed entry that deviates more than this from the recent
|
||
# median for the same pair is treated as a likely parsing error (e.g. a
|
||
# stray number from elsewhere in the message getting picked up as the
|
||
# entry) rather than a real price, and the signal is skipped.
|
||
PRICE_SANITY_PCT = 0.12
|
||
|
||
# Recognized pair/asset names that may appear right before BUY/SELL
|
||
PAIR_NAME_PATTERN = r'(GOLD|XAUUSD|XAU|SILVER|XAGUSD|XAG|BITCOIN|BTCUSD|BTC|[A-Z]{6})'
|
||
|
||
TP_DONE_RE = re.compile(r'\b(done|hit|closed)\b|✅|☑', re.IGNORECASE)
|
||
TP_NUMBER_RE = re.compile(r'\btp\s*#?\s*(\d+)\b', re.IGNORECASE)
|
||
SL_HIT_RE = re.compile(r'\bsl\s*hit\b|\bstopped\s*out\b|\bstop\s*loss\s*hit\b|\bbe\s*hit\b', re.IGNORECASE)
|
||
PARTIAL_CLOSE_RE = re.compile(r'\bclos(?:e|ing)\s+half\b|\bpartial(?:s)?\b', re.IGNORECASE)
|
||
ALL_TP_RE = re.compile(r'\ball\s+(?:take\s*profits?|tps?)\s+hit\b', re.IGNORECASE)
|
||
# Trader announcing they are moving stops to breakeven (position stays open).
|
||
# "set be", "moved to be", "sl to be", "zero risk now", "no more risk" etc.
|
||
# This is distinct from SL_HIT_RE (which means price actually hit the level).
|
||
BE_MOVE_RE = re.compile(
|
||
r'\bset\s+(?:sl\s+to\s+)?be\b' # "set be", "set sl to be"
|
||
r'|\bmov(?:e|ed|ing)\s+(?:sl\s+)?to\s+(?:be|break\s*even)\b'
|
||
r'|\bsl\s+(?:to|at)\s+(?:be|break\s*even)\b'
|
||
r'|\bzero\s+risk\b' # "zero risk now"
|
||
r'|\bno\s+more\s+risk\b', # "no more risk today"
|
||
re.IGNORECASE,
|
||
)
|
||
# Admin saying the last signal was wrong / should be ignored
|
||
CANCEL_RE = re.compile(
|
||
r'\b(?:cancel(?:led)?|wrong|mistake'
|
||
r'|ignore(?:\s+(?:last|that|this))?'
|
||
r'|don\'?t\s+take'
|
||
r'|disregard'
|
||
r'|check\s+numbers?'
|
||
r'|was\s+(?:a\s+)?(?:an?\s+)?error'
|
||
r'|bad\s+signal|false\s+signal|not\s+valid|void)\b',
|
||
re.IGNORECASE
|
||
)
|
||
|
||
|
||
|
||
def _expand_short_range_end(start, end_str):
|
||
"""Signal-Kurzschreibweise auflösen: "4126.5-29" meint 4126.5-4129,
|
||
"4198-02" meint 4198-4202. Ist das Ende schon vollständig (gleiche
|
||
Größenordnung wie der Start), wird es unverändert zurückgegeben."""
|
||
end = float(end_str)
|
||
if start <= 0 or end >= start / 10:
|
||
return end
|
||
digits = len(end_str.split('.')[0])
|
||
base = 10 ** digits
|
||
full = (int(start) // base) * base + end
|
||
if full < start - base / 2: # Überlauf, z.B. 4198-02 -> 4202
|
||
full += base
|
||
return full
|
||
|
||
|
||
class TradeSignalAnalyzer:
|
||
"""
|
||
Parses a Telegram signal export and runs an OBJECTIVE backtest against
|
||
real market data to see what a mechanical execution of the signals
|
||
would actually have produced - independent of what the provider claims
|
||
happened in the chat.
|
||
|
||
The provider's own narration (e.g. "Tp1 done", "Sl hit", "set BE") is
|
||
still captured, but only as informational metadata (`reported_*`
|
||
fields) shown next to the simulated result for comparison. It is never
|
||
used as an input to the PnL math - only real OHLC price data decides
|
||
whether a TP/SL level was actually reached.
|
||
|
||
Trade management rule that the simulation DOES bake in (because it's
|
||
described repeatedly as the provider's general method, not a one-off
|
||
claim): once price action confirms the first TP tranche, the stop for
|
||
the remaining tranches is moved to breakeven.
|
||
"""
|
||
|
||
def __init__(self, file_path, initial_capital=100, risk_level='mid',
|
||
sl_confirmation='close', compounding=False,
|
||
default_sl_pips=60, default_tp_ladder_pips=(20, 40, 60, 80),
|
||
mt5_base_url='http://localhost:8000'):
|
||
self.file_path = file_path
|
||
self.initial_capital = initial_capital
|
||
self.risk_level = risk_level
|
||
self.sl_confirmation = sl_confirmation # 'close' (default) or 'wick'
|
||
self.compounding = compounding # False (default): lot sized off initial_capital
|
||
self.default_sl_pips = default_sl_pips
|
||
self.default_tp_ladder_pips = list(default_tp_ladder_pips)
|
||
self.mt5_base_url = mt5_base_url.rstrip('/')
|
||
self.signals = []
|
||
self.trades_data = {}
|
||
self._recent_entries = {} # pair -> list of recently accepted entry prices
|
||
self.skipped_signals = [] # log of signals rejected by the sanity check
|
||
self.defaulted_signals = [] # log of informal signals that got default SL/TP/entry filled in
|
||
|
||
# ------------------------------------------------------------------
|
||
# Parsing
|
||
# ------------------------------------------------------------------
|
||
def parse_signals(self):
|
||
with open(self.file_path, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
raw_messages = re.split(r'-{10,}', content)
|
||
|
||
parsed = []
|
||
for raw in raw_messages:
|
||
if not raw.strip():
|
||
continue
|
||
match = re.match(r'Date:\s*(.*?),\s*Message:\s*(.*)', raw.strip(), re.DOTALL)
|
||
if not match:
|
||
continue
|
||
date_str, msg = match.groups()
|
||
try:
|
||
date = datetime.strptime(date_str.strip(), '%Y-%m-%d %H:%M:%S%z')
|
||
except ValueError:
|
||
continue
|
||
msg = msg.strip()
|
||
if msg:
|
||
parsed.append((date, msg))
|
||
|
||
# Telegram exports are typically newest-first; the management
|
||
# messages (TP done, SL hit, ...) only make sense relative to the
|
||
# signal that came BEFORE them in time, so we must process in
|
||
# chronological order regardless of file order.
|
||
parsed.sort(key=lambda x: x[0])
|
||
|
||
for date, msg in parsed:
|
||
self.process_message(date, msg)
|
||
|
||
self._drop_superseded_informal_signals()
|
||
|
||
def process_message(self, date, msg):
|
||
"""Handle a single message - either a new entry signal or a
|
||
management message (TP done, SL hit, partial close) against the
|
||
most recently opened signal. Used both for file replay (above)
|
||
and for a live Telegram listener processing messages one at a
|
||
time as they arrive (see live_signal_trader.py). Returns the new
|
||
signal dict if one was opened, otherwise None."""
|
||
signal = self._parse_new_trade(msg, date)
|
||
if signal:
|
||
self.signals.append(signal)
|
||
return signal
|
||
|
||
if not self.signals or self.signals[-1]['status'] != 'OPEN':
|
||
return None
|
||
|
||
if CANCEL_RE.search(msg):
|
||
# Admin retracted the last signal - mark it cancelled so the
|
||
# live trader can close any position it already opened.
|
||
self.signals[-1]['status'] = 'CANCELLED'
|
||
self.signals[-1]['cancel_date'] = date
|
||
self.signals[-1]['cancel_reason'] = msg.strip()
|
||
return 'CANCEL'
|
||
|
||
if BE_MOVE_RE.search(msg):
|
||
# Trader is moving their SL to breakeven (position stays OPEN).
|
||
# Return sentinel so the live trader can relay the SL adjustment.
|
||
self.signals[-1]['be_move_date'] = date
|
||
return 'BE_MOVE'
|
||
|
||
if SL_HIT_RE.search(msg):
|
||
self._handle_sl_hit(date, msg)
|
||
elif ALL_TP_RE.search(msg):
|
||
self._handle_tp_report(date, msg, all_levels=True)
|
||
elif TP_NUMBER_RE.search(msg) and TP_DONE_RE.search(msg):
|
||
self._handle_tp_report(date, msg)
|
||
elif PARTIAL_CLOSE_RE.search(msg):
|
||
self._handle_partial_close(date, msg)
|
||
return None
|
||
|
||
def _drop_superseded_informal_signals(self):
|
||
"""An informal 'Gold sell now' is often just a precursor to a fully
|
||
detailed message seconds/minutes later for the same setup (exactly
|
||
like your Jan 22 sample). If a detailed signal for the same pair
|
||
and direction shows up within INFORMAL_MERGE_WINDOW after an
|
||
informal one, the informal one was only the announcement - drop it
|
||
so the trade isn't counted twice."""
|
||
detailed = [s for s in self.signals if not s['is_informal']]
|
||
kept = []
|
||
for s in self.signals:
|
||
if s['is_informal'] and any(
|
||
d['pair'] == s['pair'] and d['action'] == s['action']
|
||
and 0 <= (d['date'] - s['date']).total_seconds() <= INFORMAL_MERGE_WINDOW_SECONDS
|
||
for d in detailed
|
||
):
|
||
continue
|
||
kept.append(s)
|
||
self.signals = kept
|
||
|
||
def _parse_new_trade(self, message, date):
|
||
"""Parse an entry signal. Two shapes are recognized:
|
||
|
||
1. Fully detailed, e.g.:
|
||
Gold sell 4827-4831
|
||
SL: 4835
|
||
TP: 4825
|
||
TP: 4823
|
||
2. Informal, e.g. "Gold buy now" - just a pair + direction, with no
|
||
SL/TP (sometimes no price at all). These are still returned as
|
||
tradeable signals (is_informal=True) with sl/tp_levels left as
|
||
None/[] so they can be filled in with your default risk plan
|
||
once market data is available (see _resolve_informal_signals).
|
||
"""
|
||
pair_action = re.search(
|
||
PAIR_NAME_PATTERN + r'\s+(BUY|SELL)\b', message, re.IGNORECASE
|
||
)
|
||
if not pair_action:
|
||
return None
|
||
|
||
pair_raw = pair_action.group(1).upper()
|
||
pair = PAIR_ALIASES.get(pair_raw, pair_raw)
|
||
action = pair_action.group(2).upper()
|
||
|
||
# Entry: provider usually gives a range ("4827-4831"); fall back to
|
||
# a single number right after the pair/direction if no range is
|
||
# found. Restricted to the text between "BUY/SELL" and the SL
|
||
# marker - searching the whole message risked picking up an
|
||
# unrelated number (a pip count, an aside, etc.) as the entry.
|
||
# When more than one candidate range/number shows up in that
|
||
# segment (e.g. "after a 20-30 pip pullback 2640-2643"), pick
|
||
# whichever is closest to the recent price history for this pair,
|
||
# rather than blindly taking the first match.
|
||
tail = message[pair_action.end():]
|
||
entry_segment = re.split(r'\bSL\b', tail, maxsplit=1, flags=re.IGNORECASE)[0]
|
||
|
||
range_matches = re.findall(r'(\d+(?:\.\d+)?)\s*-\s*(\d+(?:\.\d+)?)', entry_segment)
|
||
if range_matches:
|
||
candidates = []
|
||
for a, b in range_matches:
|
||
lo, hi = float(a), _expand_short_range_end(float(a), b)
|
||
candidates.append(((lo + hi) / 2, lo, hi))
|
||
else:
|
||
single_matches = re.findall(r'\d+(?:\.\d+)?', entry_segment)
|
||
candidates = [(float(x), float(x), float(x)) for x in single_matches] if single_matches else []
|
||
|
||
entry = entry_low = entry_high = None
|
||
if candidates:
|
||
entry, entry_low, entry_high = self._pick_best_candidate(pair, candidates)
|
||
|
||
# SL: there should only be one real stop-loss, but a longer message
|
||
# can mention "SL" more than once (e.g. trailing commentary after
|
||
# the TP block). No fixed ordering between SL and TP is assumed -
|
||
# collect every candidate and, once the entry is known, take the
|
||
# one closest to it rather than just the first match in the text.
|
||
# The colon is required: without it, a bare "tp1"/"sl" mention in
|
||
# casual prose (no price actually following) would otherwise let
|
||
# the regex backtrack into capturing the label digit itself.
|
||
sl_candidates = [float(x) for x in
|
||
re.findall(r'\bSL\s*\d*\s*:\s*([\d]+(?:\.\d+)?)', message, re.IGNORECASE)]
|
||
if not sl_candidates:
|
||
sl = None
|
||
elif entry is not None and len(sl_candidates) > 1:
|
||
sl = min(sl_candidates, key=lambda x: abs(x - entry))
|
||
else:
|
||
sl = sl_candidates[0]
|
||
|
||
tp_matches = re.findall(r'\bTP\s*\d*\s*:\s*(?:open\s*)?([\d]+(?:\.\d+)?)', message, re.IGNORECASE)
|
||
|
||
# Drop clearly truncated / corrupted TP values (e.g. "TP: 41" that should
|
||
# be "TP: 4170" but got cut off in the Telegram message) BEFORE the
|
||
# is_informal check so a single bad line doesn't cause the whole signal to
|
||
# be rejected by _is_sane. Any candidate that deviates more than
|
||
# PRICE_SANITY_PCT from the parsed entry is almost certainly a truncation.
|
||
if entry is not None and tp_matches:
|
||
tp_matches = [x for x in tp_matches
|
||
if abs(float(x) - entry) / entry <= PRICE_SANITY_PCT]
|
||
|
||
# Direction correction: if the price structure contradicts the text keyword
|
||
# (e.g. trader typed "Gold buy" but SL is above entry → it's a SELL), trust
|
||
# the numbers over the word. This handles typos like "my mess have a error".
|
||
# Priority: SL position > TP direction > text keyword (fallback).
|
||
if entry is not None and sl is not None:
|
||
structure_action = 'SELL' if sl > entry else 'BUY'
|
||
if structure_action != action:
|
||
import logging as _log
|
||
_log.getLogger(__name__).warning(
|
||
f"Direction mismatch: text says {action} but SL {sl} is "
|
||
f"{'above' if sl > entry else 'below'} entry {entry} "
|
||
f"→ correcting to {structure_action}"
|
||
)
|
||
action = structure_action
|
||
elif entry is not None and tp_matches:
|
||
# No SL yet - infer from TP side as a weaker signal
|
||
tp_vals = [float(x) for x in tp_matches]
|
||
tp_below = sum(1 for t in tp_vals if t < entry)
|
||
tp_above = sum(1 for t in tp_vals if t > entry)
|
||
if tp_below > tp_above and action == 'BUY':
|
||
import logging as _log
|
||
_log.getLogger(__name__).warning(
|
||
f"Direction mismatch: text says BUY but TPs are mostly below entry {entry} "
|
||
f"→ correcting to SELL"
|
||
)
|
||
action = 'SELL'
|
||
elif tp_above > tp_below and action == 'SELL':
|
||
import logging as _log
|
||
_log.getLogger(__name__).warning(
|
||
f"Direction mismatch: text says SELL but TPs are mostly above entry {entry} "
|
||
f"→ correcting to BUY"
|
||
)
|
||
action = 'BUY'
|
||
|
||
is_informal = not (sl is not None and tp_matches)
|
||
|
||
if not is_informal:
|
||
# Fully detailed signal: apply the strict checks, since a bad
|
||
# parse here silently corrupts the simulation (as you saw).
|
||
if entry is None:
|
||
return None
|
||
tp_levels = sorted({float(x) for x in tp_matches}, reverse=(action == 'SELL'))
|
||
if not self._is_sane(pair, action, entry, sl, tp_levels, date):
|
||
return None
|
||
self._recent_entries.setdefault(pair, []).append(entry)
|
||
else:
|
||
# Informal signal: SL/TP will be filled with your default risk
|
||
# plan later. An explicit entry price is only kept if it's not
|
||
# an obvious mis-parse; otherwise it's left None and resolved
|
||
# from market data (the next available price) once data is
|
||
# fetched, rather than rejecting the whole signal.
|
||
tp_levels = sorted({float(x) for x in tp_matches}, reverse=(action == 'SELL')) if tp_matches else []
|
||
if entry is not None and not self._entry_plausible(pair, entry):
|
||
entry = entry_low = entry_high = None
|
||
if entry is not None:
|
||
self._recent_entries.setdefault(pair, []).append(entry)
|
||
|
||
return {
|
||
'date': date,
|
||
'pair': pair,
|
||
'pair_raw': pair_raw,
|
||
'action': action,
|
||
'entry': entry, # may be None for informal signals - resolved from market data later
|
||
'entry_low': entry_low,
|
||
'entry_high': entry_high,
|
||
'sl': sl, # may be None for informal signals - default-filled later
|
||
'tp_levels': tp_levels, # may be [] for informal signals - default-filled later
|
||
'is_informal': is_informal,
|
||
'status': 'OPEN',
|
||
# Informational only - filled in by management messages below.
|
||
'reported_tp_hits': set(),
|
||
'reported_close_type': None,
|
||
'reported_close_date': None,
|
||
}
|
||
|
||
def _pick_best_candidate(self, pair, candidates):
|
||
"""candidates: list of (entry, low, high) tuples. Picks whichever
|
||
entry is closest to the recent price history for this pair; with
|
||
no history yet, falls back to the last candidate in the segment
|
||
(in practice the actual price tends to be stated right before the
|
||
SL line, after any incidental pip/context mentions)."""
|
||
recent = self._recent_entries.get(pair, [])
|
||
if recent:
|
||
reference = statistics.median(recent[-5:])
|
||
return min(candidates, key=lambda c: abs(c[0] - reference))
|
||
return candidates[-1]
|
||
|
||
def _entry_plausible(self, pair, entry):
|
||
recent = self._recent_entries.get(pair, [])
|
||
if len(recent) >= 3:
|
||
reference = statistics.median(recent[-5:])
|
||
return abs(entry - reference) / reference <= PRICE_SANITY_PCT
|
||
return True
|
||
|
||
def _is_sane(self, pair, action, entry, sl, tp_levels, date):
|
||
"""Reject a parsed signal if the entry looks like a mis-parse
|
||
rather than a real price (e.g. entry=3141.75 / 1351.00 showing up
|
||
among neighbors all around 2650 - a stray number got captured
|
||
instead of the actual entry), if SL/TP sit implausibly far from
|
||
the entry for what's meant to be a scalp setup, or if a TP/SL
|
||
ended up on the wrong side of entry for the trade direction (a
|
||
SELL's TP must be below entry, its SL above - otherwise "hitting
|
||
the TP" mechanically realizes a loss instead of a profit, which
|
||
is exactly how one trade in your run wiped the whole account on a
|
||
single signal despite showing "all TPs hit").
|
||
"""
|
||
recent = self._recent_entries.get(pair, [])
|
||
if len(recent) >= 3:
|
||
reference = statistics.median(recent[-5:])
|
||
if abs(entry - reference) / reference > PRICE_SANITY_PCT:
|
||
self.skipped_signals.append((date, pair, entry, 'entry deviates from recent median'))
|
||
print(f"Skipping signal at {date} ({pair}): entry {entry} is "
|
||
f">{PRICE_SANITY_PCT*100:.0f}% away from the recent median "
|
||
f"({reference:.2f}) - looks like a parsing error, check the source message.")
|
||
return False
|
||
|
||
for label, level in [('SL', sl)] + [(f'TP{i+1}', t) for i, t in enumerate(tp_levels)]:
|
||
if abs(level - entry) / entry > PRICE_SANITY_PCT:
|
||
self.skipped_signals.append((date, pair, entry, f'{label} too far from entry'))
|
||
print(f"Skipping signal at {date} ({pair}): {label}={level} is implausibly far "
|
||
f"from entry {entry} - looks like a parsing error, check the source message.")
|
||
return False
|
||
|
||
sl_wrong_side = (action == 'BUY' and sl >= entry) or (action == 'SELL' and sl <= entry)
|
||
if sl_wrong_side:
|
||
self.skipped_signals.append((date, pair, entry, 'SL on the wrong side of entry'))
|
||
print(f"Skipping signal at {date} ({pair}): SL={sl} is on the wrong side of "
|
||
f"entry {entry} for a {action} - looks like a parsing error, check the source message.")
|
||
return False
|
||
|
||
tp_wrong_side = any((action == 'BUY' and tp <= entry) or (action == 'SELL' and tp >= entry)
|
||
for tp in tp_levels)
|
||
if tp_wrong_side:
|
||
self.skipped_signals.append((date, pair, entry, 'a TP is on the wrong side of entry'))
|
||
print(f"Skipping signal at {date} ({pair}): one of {tp_levels} is on the wrong side of "
|
||
f"entry {entry} for a {action} - looks like a parsing error, check the source message.")
|
||
return False
|
||
|
||
return True
|
||
|
||
def _handle_sl_hit(self, date, msg):
|
||
sig = self.signals[-1]
|
||
sig['status'] = 'CLOSED'
|
||
sig['reported_close_type'] = 'BE' if 'be hit' in msg.lower() else 'SL'
|
||
sig['reported_close_date'] = date
|
||
|
||
def _handle_tp_report(self, date, msg, all_levels=False):
|
||
sig = self.signals[-1]
|
||
if all_levels:
|
||
sig['reported_tp_hits'].update(range(len(sig['tp_levels'])))
|
||
else:
|
||
for n in TP_NUMBER_RE.findall(msg):
|
||
idx = int(n) - 1
|
||
if 0 <= idx < len(sig['tp_levels']):
|
||
sig['reported_tp_hits'].add(idx)
|
||
if len(sig['reported_tp_hits']) >= len(sig['tp_levels']) and sig['tp_levels']:
|
||
sig['status'] = 'CLOSED'
|
||
sig['reported_close_type'] = 'ALL_TP'
|
||
sig['reported_close_date'] = date
|
||
|
||
def _handle_partial_close(self, date, msg):
|
||
sig = self.signals[-1]
|
||
sig['reported_partial_close_date'] = date
|
||
|
||
# ------------------------------------------------------------------
|
||
# Market data
|
||
# ------------------------------------------------------------------
|
||
def fetch_forex_data(self):
|
||
unique_pairs = {s['pair'] for s in self.signals}
|
||
|
||
for pair in unique_pairs:
|
||
symbol = MT5_SYMBOL_OVERRIDES.get(pair, pair)
|
||
dates = [s['date'].replace(tzinfo=None) for s in self.signals if s['pair'] == pair]
|
||
earliest_needed = min(dates)
|
||
|
||
df = None
|
||
for tf in MT5_TIMEFRAME_FALLBACK_CHAIN:
|
||
candidate = self._mt5_fetch_ohlc(symbol, tf)
|
||
if candidate is None or candidate.empty:
|
||
continue
|
||
if candidate.index.min() <= earliest_needed:
|
||
df = candidate
|
||
print(f"{pair} ({symbol}): using {tf} bars, "
|
||
f"{candidate.index.min()} -> {candidate.index.max()} ({len(candidate)} bars)")
|
||
break
|
||
print(f"{pair} ({symbol}): {tf} only reaches back to {candidate.index.min()}, "
|
||
f"need {earliest_needed} - trying a coarser timeframe")
|
||
df = candidate # keep the coarsest successful fetch as a fallback
|
||
|
||
if df is None:
|
||
print(f"No MT5 data available for {pair} ({symbol}) - check the symbol name "
|
||
f"via GET {self.mt5_base_url}/symbols")
|
||
self.trades_data[pair] = df
|
||
|
||
self._resolve_informal_signals()
|
||
|
||
def _resolve_informal_signals(self):
|
||
"""Fill in whatever an informal signal ('Gold buy now') didn't
|
||
specify: entry, if no price was in the message, comes from the
|
||
first available market price at/after the signal time (a market-
|
||
order fill); SL/TP, if missing, come from your default risk plan
|
||
(default_sl_pips / default_tp_ladder_pips), centered on whichever
|
||
entry ended up being used (explicit or market-resolved)."""
|
||
for sig in self.signals:
|
||
if not sig['is_informal']:
|
||
continue
|
||
|
||
data = self.trades_data.get(sig['pair'])
|
||
|
||
if sig['entry'] is None:
|
||
if data is None or data.empty:
|
||
continue # no entry and no data - will show as NO_DATA downstream
|
||
signal_date = sig['date'].replace(tzinfo=None)
|
||
future = data[data.index >= signal_date]
|
||
if future.empty:
|
||
continue
|
||
sig['entry'] = float(future.iloc[0]['Open'])
|
||
sig['entry_low'] = sig['entry_high'] = sig['entry']
|
||
self._recent_entries.setdefault(sig['pair'], []).append(sig['entry'])
|
||
self.defaulted_signals.append(
|
||
(sig['date'], sig['pair'], sig['action'], 'entry',
|
||
f"no price in message, used market open {sig['entry']:.2f}"))
|
||
|
||
pip = PIP_SIZE.get(sig['pair'], DEFAULT_PIP_SIZE)
|
||
|
||
if sig['sl'] is None:
|
||
distance = self.default_sl_pips * pip
|
||
sig['sl'] = sig['entry'] - distance if sig['action'] == 'BUY' else sig['entry'] + distance
|
||
self.defaulted_signals.append(
|
||
(sig['date'], sig['pair'], sig['action'], 'SL',
|
||
f"no SL in message, defaulted to {self.default_sl_pips} pips ({sig['sl']:.2f})"))
|
||
|
||
if not sig['tp_levels']:
|
||
levels = [sig['entry'] + (d * pip if sig['action'] == 'BUY' else -d * pip)
|
||
for d in self.default_tp_ladder_pips]
|
||
sig['tp_levels'] = sorted(levels, reverse=(sig['action'] == 'SELL'))
|
||
self.defaulted_signals.append(
|
||
(sig['date'], sig['pair'], sig['action'], 'TP',
|
||
f"no TP in message, defaulted to {self.default_tp_ladder_pips} pip ladder"))
|
||
|
||
def _mt5_fetch_ohlc(self, symbol, timeframe, bars=10000):
|
||
"""Pull OHLC bars from your MT5 MCP server's /ohlc endpoint."""
|
||
url = f"{self.mt5_base_url}/ohlc/{symbol}"
|
||
try:
|
||
resp = requests.get(url, params={'timeframe': timeframe, 'bars': bars}, timeout=30)
|
||
resp.raise_for_status()
|
||
payload = resp.json()
|
||
except Exception as e:
|
||
print(f"MT5 MCP fetch failed for {symbol} {timeframe}: {e}")
|
||
return None
|
||
|
||
rows = payload.get('data', [])
|
||
if not rows:
|
||
return None
|
||
|
||
df = pd.DataFrame(rows)
|
||
df['time'] = pd.to_datetime(df['time'], utc=True).dt.tz_localize(None)
|
||
df = df.set_index('time').sort_index()
|
||
df = df.rename(columns={
|
||
'open': 'Open', 'high': 'High', 'low': 'Low',
|
||
'close': 'Close', 'volume': 'Volume',
|
||
})
|
||
return df
|
||
|
||
# ------------------------------------------------------------------
|
||
# Backtest
|
||
# ------------------------------------------------------------------
|
||
def analyze_trades(self, risk_level=None, sl_confirmation=None, compounding=None,
|
||
sizing_mode='fixed', min_lot=MIN_LOT):
|
||
"""Run a bar-by-bar backtest over all parsed signals.
|
||
|
||
sizing_mode options:
|
||
'fixed' - lot = (capital / 1000) * RISK_TIERS[risk_level]
|
||
'risk_pct' - lot sized so a full stop-out costs exactly
|
||
RISK_PCT_TIERS[risk_level] of capital
|
||
'survival' - lot sized so THREE full stop-outs would exhaust the
|
||
current capital (risk_pct = 1/3). If the resulting
|
||
lot < min_lot, the signal is traded with min_lot at
|
||
TP1 only (the same "small-account" fallback the live
|
||
trader uses). This is the most realistic mode for a
|
||
$100 starting balance.
|
||
"""
|
||
risk_level = risk_level or self.risk_level
|
||
sl_confirmation = sl_confirmation or self.sl_confirmation
|
||
compounding = self.compounding if compounding is None else compounding
|
||
results = []
|
||
current_capital = self.initial_capital
|
||
max_capital = current_capital
|
||
max_drawdown = 0.0
|
||
position_open_until = None # bar timestamp when the last trade closed
|
||
|
||
for signal in self.signals:
|
||
pair = signal['pair']
|
||
data = self.trades_data.get(pair)
|
||
|
||
row = {
|
||
'date': signal['date'],
|
||
'pair': pair,
|
||
'action': signal['action'],
|
||
'entry': signal['entry'],
|
||
'sl': signal['sl'],
|
||
'tp_count': len(signal['tp_levels']),
|
||
'signal_type': 'informal' if signal['is_informal'] else 'detailed',
|
||
'reported_result': self._reported_result_label(signal),
|
||
}
|
||
|
||
if signal['entry'] is None or data is None or data.empty:
|
||
row.update(result='NO_DATA', pnl=0.0, pnl_pct=0.0, capital=current_capital,
|
||
drawdown=max_drawdown, sizing='NO_DATA')
|
||
results.append(row)
|
||
continue
|
||
|
||
signal_date = signal['date'].replace(tzinfo=None)
|
||
|
||
# Skip signals that arrive while a previous trade is still open —
|
||
# mirrors the live trader's max_concurrent=1 guard exactly.
|
||
if position_open_until is not None and signal_date <= position_open_until:
|
||
row.update(result='SKIPPED(concurrent)', pnl=0.0, pnl_pct=0.0,
|
||
capital=current_capital, drawdown=max_drawdown, lot=0.0,
|
||
sizing='blocked')
|
||
results.append(row)
|
||
continue
|
||
|
||
trade_data = data[data.index >= signal_date]
|
||
if trade_data.empty:
|
||
row.update(result='NO_DATA', pnl=0.0, pnl_pct=0.0, capital=current_capital,
|
||
drawdown=max_drawdown, sizing='NO_DATA')
|
||
results.append(row)
|
||
continue
|
||
|
||
sizing_capital = current_capital if compounding else self.initial_capital
|
||
|
||
if sizing_mode == 'survival':
|
||
# 3-signal survival rule (mirrors the live trader exactly):
|
||
# lot that costs balance/3 on a full stop-out
|
||
survival_lot = risk_based_lot(sizing_capital, 1.0 / 3.0,
|
||
signal['entry'], signal['sl'], pair)
|
||
if survival_lot < min_lot:
|
||
# Too small for the full ladder - use min_lot / TP1 only
|
||
lot = min_lot
|
||
effective_tp_levels = signal['tp_levels'][:1]
|
||
sim_signal = dict(signal, tp_levels=effective_tp_levels)
|
||
sizing_tag = f'survival/TP1({min_lot})'
|
||
else:
|
||
lot = survival_lot
|
||
sim_signal = signal
|
||
sizing_tag = f'survival({lot:.4f})'
|
||
elif sizing_mode == 'risk_pct':
|
||
lot = risk_based_lot(sizing_capital, RISK_PCT_TIERS[risk_level],
|
||
signal['entry'], signal['sl'], pair)
|
||
sim_signal = signal
|
||
sizing_tag = f'risk_pct/{risk_level}({lot:.4f})'
|
||
else:
|
||
lot = (sizing_capital / 1000.0) * RISK_TIERS[risk_level]
|
||
sim_signal = signal
|
||
sizing_tag = f'fixed/{risk_level}({lot:.4f})'
|
||
|
||
pnl, sim_result, close_time = self._simulate_trade(sim_signal, trade_data, lot, sl_confirmation)
|
||
|
||
# Record when this position closed so subsequent signals can be blocked.
|
||
if close_time is not None:
|
||
position_open_until = close_time
|
||
|
||
capital_before = current_capital
|
||
current_capital = max(0.0, current_capital + pnl)
|
||
max_capital = max(max_capital, current_capital)
|
||
drawdown = (max_capital - current_capital) / max_capital * 100 if max_capital > 0 else 0.0
|
||
max_drawdown = max(max_drawdown, drawdown)
|
||
pnl_pct = (pnl / capital_before * 100) if capital_before else 0.0
|
||
|
||
row.update(result=sim_result, pnl=pnl, pnl_pct=pnl_pct, capital=current_capital,
|
||
drawdown=drawdown, lot=lot, sizing=sizing_tag)
|
||
results.append(row)
|
||
|
||
return pd.DataFrame(results)
|
||
|
||
def _simulate_trade(self, signal, trade_data, lot, sl_confirmation):
|
||
"""Bar-by-bar simulation against real OHLC data.
|
||
|
||
Position size is FIXED at trade-open time: lot (sized off current
|
||
equity via your risk-tier table) x the instrument's contract size,
|
||
split evenly across one tranche per TP level. This replaces an
|
||
earlier leverage*equity model that re-leveraged the whole, already
|
||
compounded balance into every trade and blew up exponentially.
|
||
|
||
TP fills use the bar's High/Low (a touch is a real limit fill).
|
||
SL/breakeven exits use the bar's Close by default (sl_confirmation
|
||
='close') rather than the wick, mirroring your stated approach of
|
||
not hard-placing the stop so a brief spread spike doesn't take you
|
||
out. Pass sl_confirmation='wick' for the stricter, literal version.
|
||
"""
|
||
action = signal['action']
|
||
entry_price = signal['entry']
|
||
tp_levels = signal['tp_levels']
|
||
n_tranches = max(len(tp_levels), 1)
|
||
|
||
contract_size = CONTRACT_SIZE.get(signal['pair'], DEFAULT_CONTRACT_SIZE)
|
||
dollar_per_unit = lot * contract_size
|
||
tranche_dollar_per_unit = dollar_per_unit / n_tranches
|
||
|
||
current_sl = signal['sl']
|
||
remaining = list(range(len(tp_levels)))
|
||
pnl = 0.0
|
||
hit_labels = []
|
||
|
||
close_bar_time = None
|
||
for bar_time, bar in trade_data.iterrows():
|
||
close_bar_time = bar_time
|
||
high, low, close = float(bar['High']), float(bar['Low']), float(bar['Close'])
|
||
sl_ref = close if sl_confirmation == 'close' else (low if action == 'BUY' else high)
|
||
|
||
sl_triggered = (action == 'BUY' and sl_ref <= current_sl) or \
|
||
(action == 'SELL' and sl_ref >= current_sl)
|
||
if sl_triggered and remaining:
|
||
for _ in remaining:
|
||
if action == 'BUY':
|
||
pnl += tranche_dollar_per_unit * (current_sl - entry_price)
|
||
else:
|
||
pnl += tranche_dollar_per_unit * (entry_price - current_sl)
|
||
hit_labels.append('BE' if current_sl == entry_price else 'SL')
|
||
remaining = []
|
||
break
|
||
|
||
for idx in list(remaining):
|
||
tp_price = tp_levels[idx]
|
||
hit = (action == 'BUY' and high >= tp_price) or \
|
||
(action == 'SELL' and low <= tp_price)
|
||
if hit:
|
||
if action == 'BUY':
|
||
pnl += tranche_dollar_per_unit * (tp_price - entry_price)
|
||
else:
|
||
pnl += tranche_dollar_per_unit * (entry_price - tp_price)
|
||
remaining.remove(idx)
|
||
hit_labels.append(f'TP{idx + 1}')
|
||
if current_sl != entry_price:
|
||
current_sl = entry_price # breakeven after first confirmed TP
|
||
|
||
if not remaining:
|
||
break
|
||
|
||
if remaining:
|
||
last_close = float(trade_data.iloc[-1]['Close'])
|
||
for _ in remaining:
|
||
if action == 'BUY':
|
||
pnl += tranche_dollar_per_unit * (last_close - entry_price)
|
||
else:
|
||
pnl += tranche_dollar_per_unit * (entry_price - last_close)
|
||
hit_labels.append('OPEN(MTM)')
|
||
|
||
return pnl, '+'.join(hit_labels) if hit_labels else 'UNKNOWN', close_bar_time
|
||
|
||
@staticmethod
|
||
def _reported_result_label(signal):
|
||
if signal['reported_close_type'] == 'ALL_TP':
|
||
return f"ALL_TP({len(signal['tp_levels'])})"
|
||
if signal['reported_close_type']:
|
||
tps = ','.join(f'TP{i+1}' for i in sorted(signal['reported_tp_hits']))
|
||
return f"{tps}+{signal['reported_close_type']}" if tps else signal['reported_close_type']
|
||
if signal['reported_tp_hits']:
|
||
return ','.join(f'TP{i+1}' for i in sorted(signal['reported_tp_hits']))
|
||
return 'OPEN'
|
||
|
||
|
||
def _fetch_from_telegram(analyzer, channel_id, days):
|
||
"""Pull `days` days of history from a Telegram channel and feed each
|
||
message through analyzer.process_message(). Returns (total_msgs, skipped).
|
||
|
||
Requires TELEGRAM_API_ID / TELEGRAM_API_HASH / PHONE_NUMBER env vars.
|
||
Uses a temporary session file so it never conflicts with the live trader.
|
||
"""
|
||
import os
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
from datetime import datetime, timezone, timedelta
|
||
from telethon.sync import TelegramClient
|
||
|
||
api_id = int(os.environ['API_ID'])
|
||
api_hash = os.environ['API_HASH']
|
||
phone = os.environ['PHONE_NUMBER']
|
||
|
||
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
|
||
print(f"Connecting to Telegram, fetching {days} days of history "
|
||
f"(since {cutoff.date()}) from channel {channel_id} …")
|
||
|
||
total_msgs = 0
|
||
skipped = 0
|
||
# Reuse the live trader's existing session so no re-authentication is needed.
|
||
session_name = 'mysession'
|
||
|
||
with TelegramClient(session_name, api_id, api_hash) as client:
|
||
client.start(phone=phone)
|
||
|
||
# Resolve the channel by walking iter_dialogs() — this gives us the
|
||
# fully resolved entity (with access hash) regardless of session cache.
|
||
bare_id = abs(channel_id)
|
||
if str(bare_id).startswith('100') and len(str(bare_id)) > 10:
|
||
bare_id = int(str(bare_id)[3:])
|
||
|
||
entity = None
|
||
print("Resolving channel from dialogs…")
|
||
for dialog in client.iter_dialogs():
|
||
if getattr(dialog.entity, 'id', None) == bare_id:
|
||
entity = dialog.entity
|
||
break
|
||
|
||
if entity is None:
|
||
raise ValueError(
|
||
f"Channel {channel_id} not found in your dialogs. "
|
||
"Make sure you are a member of the channel."
|
||
)
|
||
|
||
for msg in client.iter_messages(entity, reverse=True, offset_date=cutoff):
|
||
text = (msg.text or '').strip()
|
||
if not text:
|
||
skipped += 1
|
||
continue
|
||
total_msgs += 1
|
||
analyzer.process_message(msg.date, text)
|
||
|
||
analyzer._drop_superseded_informal_signals()
|
||
return total_msgs, skipped
|
||
|
||
|
||
def main():
|
||
"""Standalone backtester for the Goldex channel export.
|
||
|
||
File mode (default):
|
||
python3 traderanalyser.py <file> [mt5_url] [tier] [compound] [capital] [sizing]
|
||
|
||
Telegram mode (live history fetch – no file needed):
|
||
python3 traderanalyser.py --telegram [channel_id] [days] [mt5_url] [tier] [compound] [capital] [sizing]
|
||
|
||
In Telegram mode the script connects using TELEGRAM_API_ID / TELEGRAM_API_HASH /
|
||
PHONE_NUMBER env vars, fetches the last N days (default 5) of history from
|
||
the Goldex channel, feeds every message through the parser, then runs the
|
||
full backtest — identical output to file mode.
|
||
|
||
channel_id defaults to the CHANNEL_ID env var.
|
||
"""
|
||
import sys
|
||
import os
|
||
|
||
# ── Telegram mode ────────────────────────────────────────────────────────
|
||
if len(sys.argv) > 1 and sys.argv[1] == '--telegram':
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
argv = sys.argv[2:]
|
||
channel_id = int(argv[0]) if len(argv) > 0 else int(os.environ.get('CHANNEL_ID', 0))
|
||
days = int(argv[1]) if len(argv) > 1 else 5
|
||
mt5_base_url = argv[2] if len(argv) > 2 else 'http://localhost:8000'
|
||
chosen_tier = argv[3] if len(argv) > 3 else 'mid'
|
||
compounding = argv[4].lower() in ('1', 'true', 'yes', 'compound') if len(argv) > 4 else False
|
||
initial_capital = float(argv[5]) if len(argv) > 5 else 100
|
||
sizing_mode = argv[6] if len(argv) > 6 else 'survival'
|
||
|
||
if not channel_id:
|
||
print("ERROR: channel_id required (pass as arg or set CHANNEL_ID env var).")
|
||
sys.exit(1)
|
||
|
||
analyzer = TradeSignalAnalyzer(
|
||
file_path='<telegram>',
|
||
initial_capital=initial_capital,
|
||
compounding=compounding,
|
||
mt5_base_url=mt5_base_url,
|
||
)
|
||
total_msgs, skipped = _fetch_from_telegram(analyzer, channel_id, days)
|
||
print(f"Fetched {total_msgs} messages ({skipped} empty/media skipped) "
|
||
f"from channel {channel_id} over the last {days} days.")
|
||
|
||
# ── File mode ────────────────────────────────────────────────────────────
|
||
else:
|
||
file_path = sys.argv[1] if len(sys.argv) > 1 else 'telegram_messages.txt'
|
||
mt5_base_url = sys.argv[2] if len(sys.argv) > 2 else 'http://localhost:8000'
|
||
chosen_tier = sys.argv[3] if len(sys.argv) > 3 else 'mid'
|
||
compounding = (sys.argv[4].lower() in ('1', 'true', 'yes', 'compound')
|
||
if len(sys.argv) > 4 else False)
|
||
initial_capital = float(sys.argv[5]) if len(sys.argv) > 5 else 100
|
||
sizing_mode = sys.argv[6] if len(sys.argv) > 6 else 'survival'
|
||
|
||
analyzer = TradeSignalAnalyzer(
|
||
file_path, initial_capital=initial_capital,
|
||
compounding=compounding, mt5_base_url=mt5_base_url,
|
||
)
|
||
analyzer.parse_signals()
|
||
|
||
if not analyzer.signals:
|
||
print("No complete trade signals (pair + direction + SL + TP) were found.")
|
||
return
|
||
|
||
if analyzer.skipped_signals:
|
||
print(f"\n{len(analyzer.skipped_signals)} signal(s) skipped by the sanity check "
|
||
f"(likely parsing errors - worth checking the source messages):")
|
||
for date, pair, entry, reason in analyzer.skipped_signals:
|
||
print(f" {date} {pair} entry={entry} ({reason})")
|
||
|
||
n_informal = sum(1 for s in analyzer.signals if s['is_informal'])
|
||
if n_informal:
|
||
print(f"\n{n_informal} informal signal(s) (e.g. \"Gold buy now\") kept for trading "
|
||
f"with defaults filled in:")
|
||
|
||
analyzer.fetch_forex_data()
|
||
|
||
if analyzer.defaulted_signals:
|
||
for date, pair, action, field, note in analyzer.defaulted_signals:
|
||
print(f" {date} {pair} {action}: {note}")
|
||
|
||
mode_label = 'COMPOUNDING (lot grows with balance)' if compounding else 'FIXED (lot sized off initial capital only)'
|
||
print(f"\nRisk tier comparison [{mode_label}] "
|
||
f"- pass 'compound' as a 4th argument to switch modes:")
|
||
print("-" * 70)
|
||
print(" FIXED table (lot per $1000 of capital - ignores stop width):")
|
||
for tier in ('high', 'mid', 'low'):
|
||
df = analyzer.analyze_trades(risk_level=tier, sizing_mode='fixed')
|
||
evaluated = df[df['result'] != 'NO_DATA']
|
||
final_capital = df['capital'].iloc[-1] if not df.empty else analyzer.initial_capital
|
||
total_return = (final_capital - analyzer.initial_capital) / analyzer.initial_capital * 100
|
||
win_rate = evaluated['pnl'].gt(0).mean() * 100 if not evaluated.empty else 0.0
|
||
print(f" {tier:>5} (lot={RISK_TIERS[tier]:.2f}/$1000): "
|
||
f"final capital {final_capital:,.2f} USDT, return {total_return:+.1f}%, "
|
||
f"max drawdown {df['drawdown'].max():.1f}%, win rate {win_rate:.1f}%")
|
||
|
||
print(" RISK_PCT table (risk % of capital per trade, scaled to each trade's actual stop):")
|
||
risk_pct_results = {}
|
||
for tier in ('high', 'mid', 'low'):
|
||
df = analyzer.analyze_trades(risk_level=tier, sizing_mode='risk_pct')
|
||
risk_pct_results[tier] = df
|
||
evaluated = df[df['result'] != 'NO_DATA']
|
||
final_capital = df['capital'].iloc[-1] if not df.empty else analyzer.initial_capital
|
||
total_return = (final_capital - analyzer.initial_capital) / analyzer.initial_capital * 100
|
||
win_rate = evaluated['pnl'].gt(0).mean() * 100 if not evaluated.empty else 0.0
|
||
print(f" {tier:>5} (risk={RISK_PCT_TIERS[tier]*100:.1f}%/trade): "
|
||
f"final capital {final_capital:,.2f} USDT, return {total_return:+.1f}%, "
|
||
f"max drawdown {df['drawdown'].max():.1f}%, win rate {win_rate:.1f}%")
|
||
|
||
print(" SURVIVAL table (risk 1/3 of balance per trade; if survival lot < 0.01 → TP1 only):")
|
||
df_surv = analyzer.analyze_trades(sizing_mode='survival', compounding=True)
|
||
evaluated_surv = df_surv[df_surv['result'] != 'NO_DATA']
|
||
final_surv = df_surv['capital'].iloc[-1] if not df_surv.empty else analyzer.initial_capital
|
||
total_return_surv = (final_surv - analyzer.initial_capital) / analyzer.initial_capital * 100
|
||
win_rate_surv = evaluated_surv['pnl'].gt(0).mean() * 100 if not evaluated_surv.empty else 0.0
|
||
n_survival_mode = (df_surv['sizing'].str.startswith('survival/TP1')).sum() if 'sizing' in df_surv.columns else '?'
|
||
print(f" (compounding, 1/3 risk): "
|
||
f"final capital {final_surv:,.2f} USDT, return {total_return_surv:+.1f}%, "
|
||
f"max drawdown {df_surv['drawdown'].max():.1f}%, win rate {win_rate_surv:.1f}% "
|
||
f"({n_survival_mode} trades in TP1-only fallback mode)")
|
||
|
||
results_df = analyzer.analyze_trades(risk_level=chosen_tier, sizing_mode=sizing_mode,
|
||
compounding=compounding)
|
||
|
||
if results_df.empty or 'capital' not in results_df.columns:
|
||
print("No trades could be evaluated (no signals or no market data available).")
|
||
return
|
||
|
||
tier_desc = '' if sizing_mode == 'survival' else f"'{chosen_tier}' tier, "
|
||
print(f"\nDetailed results below use {tier_desc}'{sizing_mode}' sizing "
|
||
f"(pass high/mid/low as 3rd arg, 'fixed'/'risk_pct'/'survival' as 6th, to switch):")
|
||
print("-" * 50)
|
||
print(f"Initial Capital: {analyzer.initial_capital} USDT")
|
||
print(f"Final Capital: {results_df['capital'].iloc[-1]:.2f} USDT")
|
||
print(f"Total Return: {((results_df['capital'].iloc[-1] - analyzer.initial_capital) / analyzer.initial_capital * 100):.2f}%")
|
||
print(f"Maximum Drawdown: {results_df['drawdown'].max():.2f}%")
|
||
|
||
evaluated = results_df[results_df['result'] != 'NO_DATA']
|
||
print(f"Total Signals: {len(results_df)} (No data: {(results_df['result'] == 'NO_DATA').sum()})")
|
||
if not evaluated.empty:
|
||
win_rate = evaluated['pnl'].gt(0).mean() * 100
|
||
print(f"Win Rate (simulated): {win_rate:.2f}%")
|
||
print(f"Best trade: {evaluated['pnl'].max():.2f} USDT")
|
||
print(f"Worst trade: {evaluated['pnl'].min():.2f} USDT")
|
||
print(f"Average pnl/trade: {evaluated['pnl'].mean():.2f} USDT")
|
||
if sizing_mode == 'survival' and 'sizing' in results_df.columns:
|
||
n_tp1 = results_df['sizing'].str.startswith('survival/TP1').sum()
|
||
print(f"Trades in TP1-only fallback (balance too small for ladder): {n_tp1}")
|
||
|
||
results_df.to_csv('trading_results.csv', index=False)
|
||
print("\nFull per-trade results saved to trading_results.csv")
|
||
|
||
print("\nDetailed Trade Results (per-trade profit in 'pnl' / 'pnl_pct'):")
|
||
base_cols = ['date', 'pair', 'action', 'entry', 'signal_type', 'lot', 'reported_result', 'result', 'pnl', 'pnl_pct', 'capital']
|
||
extra_cols = ['sizing'] if sizing_mode == 'survival' and 'sizing' in results_df.columns else []
|
||
cols = base_cols + extra_cols
|
||
display_df = results_df[[c for c in cols if c in results_df.columns]].copy()
|
||
if 'lot' in display_df.columns:
|
||
display_df['lot'] = display_df['lot'].round(4)
|
||
display_df['pnl'] = display_df['pnl'].round(2)
|
||
display_df['pnl_pct'] = display_df['pnl_pct'].round(2)
|
||
display_df['capital'] = display_df['capital'].round(2)
|
||
print(display_df.to_string())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|