""" 02_close_forensics.py PURPOSE ------- For every USDJPY trade closed early (before TP, before SL, before 24h hold), extract the complete event timeline from live_monitor_v2.log and identify what triggered the close. WHY THIS MATTERS ---------------- USDJPY documented exit rules: TP 0.70%, SL 0.40%, 24h max hold, NO z-exit. If trades are closing for any other reason, that's a bug — and bugs compound across every future trade. INPUTS ------ - Path to live_monitor_v2.log - List of MT5 ticket IDs to investigate - Optional: state_usdjpy.json for cross-reference USAGE ----- python 02_close_forensics.py path\\to\\live_monitor_v2.log Then enter each ticket ID when prompted. OUTPUT ------ For each ticket: - Full event timeline (signal -> arm -> entry -> all checks -> close) - Closure trigger identification - Any anomalous events (z-exit firing on UJ, watchdog interventions, etc.) """ import re import sys import json from pathlib import Path from collections import defaultdict # Regex patterns for log events we care about PATTERNS = { "signal": re.compile(r"\[USDJPY\].*SIGNAL DETECTED.*z=([-\d.]+).*dir=(\w+).*target=([\d.]+)"), "armed": re.compile(r"\[USDJPY\].*Armed.*dir=(\w+).*target=([\d.]+)"), "disarmed": re.compile(r"\[USDJPY\].*Disarmed"), "order": re.compile(r"\[USDJPY\].*[Oo]rder.*placed|filled|opened"), "entry": re.compile(r"\[USDJPY\].*[Ee]ntry.*?([\d.]+)"), "tp_set": re.compile(r"\[USDJPY\].*TP.*?([\d.]+)"), "sl_set": re.compile(r"\[USDJPY\].*SL.*?([\d.]+)"), "zscore": re.compile(r"\[USDJPY\].*Z-score:\s*([-\d.]+)"), "close_tp": re.compile(r"\[USDJPY\].*close.*?[Tt]ake.[Pp]rofit|TP hit|TP reached"), "close_sl": re.compile(r"\[USDJPY\].*close.*?[Ss]top.[Ll]oss|SL hit|stopped out"), "close_hold": re.compile(r"\[USDJPY\].*close.*?[Hh]old|24h.*expir|time.*exit"), "close_zexit":re.compile(r"\[USDJPY\].*close.*?[Zz].?exit|z-exit"), "close_any": re.compile(r"\[USDJPY\].*[Cc]los"), "watchdog": re.compile(r"\[WATCHDOG\].*USDJPY|orphan|stale|disconnect"), "error": re.compile(r"\[USDJPY\].*ERROR|WARNING|EXCEPTION", re.IGNORECASE), "ticket": re.compile(r"ticket[=:\s]*(\d+)", re.IGNORECASE), "outside": re.compile(r"\[USDJPY\].*Outside.*window|out of session"), "session": re.compile(r"session.*?(\d{2}:\d{2}).*?-.*?(\d{2}:\d{2})"), } TIMESTAMP = re.compile(r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})") def parse_log(log_path): """Read log file, return list of (timestamp, full_line) tuples.""" if not Path(log_path).exists(): print(f"ERROR: log file not found: {log_path}") sys.exit(1) entries = [] with open(log_path, encoding="utf-8", errors="replace") as f: for line in f: line = line.rstrip() m = TIMESTAMP.match(line) if m: entries.append((m.group(1), line)) print(f"Parsed {len(entries)} timestamped log entries") return entries def find_trade_window(entries, ticket): """ Locate the rough log window for a given ticket. Strategy: find any line mentioning the ticket; show ±50 lines around each. If ticket not found, prompt user for date range. """ matches = [i for i, (_, line) in enumerate(entries) if str(ticket) in line] if matches: print(f"\nTicket {ticket} appears in {len(matches)} log lines.") return matches else: print(f"\nTicket {ticket} NOT FOUND in log.") print("This is suspicious — the live monitor should log every order.") return [] def analyse_ticket(entries, ticket): """Build event timeline and flag closure cause for a single ticket.""" print("\n" + "=" * 70) print(f"INVESTIGATING TICKET {ticket}") print("=" * 70) # Find all log lines that mention this ticket ticket_lines = [(i, ts, line) for i, (ts, line) in enumerate(entries) if str(ticket) in line] if not ticket_lines: # Ticket not in log directly — prompt for date print("Ticket ID not in log. Searching by date range instead.") date = input(f"Enter trade date (YYYY-MM-DD) for ticket {ticket}: ").strip() ticket_lines = [(i, ts, line) for i, (ts, line) in enumerate(entries) if ts.startswith(date) and "USDJPY" in line] if not ticket_lines: print(f"No USDJPY entries found for {date}.") return # Print all log lines with USDJPY for the relevant time window first_idx = ticket_lines[0][0] last_idx = ticket_lines[-1][0] # Widen window by 30 lines either side start = max(0, first_idx - 30) end = min(len(entries), last_idx + 50) print(f"\nLog window: lines {start} - {end}") print(f"Time range: {entries[start][0]} to {entries[end-1][0]}\n") events = { "signal": None, "armed": None, "entry": None, "tp_set": None, "sl_set": None, "first_zscore": None, "last_zscore": None, "close_event": None, "anomalies": [], } print("FULL TIMELINE (USDJPY events only):") print("-" * 70) for i in range(start, end): ts, line = entries[i] if "USDJPY" not in line and "WATCHDOG" not in line: continue # Print all USDJPY-related lines print(f" {ts} {line[24:]}") # strip timestamp prefix for readability # Categorise for key, pat in PATTERNS.items(): if pat.search(line): if key == "signal" and events["signal"] is None: events["signal"] = (ts, line) elif key == "armed" and events["armed"] is None: events["armed"] = (ts, line) elif key == "entry" and events["entry"] is None: events["entry"] = (ts, line) elif key.startswith("close_") and events["close_event"] is None: events["close_event"] = (key, ts, line) elif key == "watchdog": events["anomalies"].append(("watchdog", ts, line)) elif key == "error": events["anomalies"].append(("error", ts, line)) # Diagnosis print("\n" + "-" * 70) print("CLOSURE DIAGNOSIS") print("-" * 70) if events["close_event"] is None: print("⚠ No explicit close event found in log — this itself is suspicious.") print(" Either the close happened outside the monitor's awareness") print(" (broker forced close, manual MT5 close) or the close logging") print(" format doesn't match the patterns we searched for.") else: kind, ts, line = events["close_event"] kind_clean = kind.replace("close_", "") print(f"Closure type detected: {kind_clean.upper()}") print(f"Closure time: {ts}") print(f"Log line: {line.strip()}") if kind == "close_zexit": print("\n⚠ ANOMALY: z-exit fired on USDJPY.") print(" Doc Section 15: USDJPY z-exit is OFF — should never fire.") print(" This is a bug in live_signal_monitor_v2.py.") elif kind == "close_any" and not any( kind.startswith(f"close_{x}") for x in ("tp", "sl", "hold") ): print("\n⚠ ANOMALY: close detected but type unknown.") print(" Closure didn't match TP/SL/hold patterns. Check log line above.") if events["anomalies"]: print("\nADDITIONAL ANOMALIES:") for kind, ts, line in events["anomalies"]: print(f" [{kind.upper()}] {ts} | {line.strip()[:120]}") def main(): if len(sys.argv) < 2: print(__doc__) sys.exit(1) log_path = sys.argv[1] entries = parse_log(log_path) print("\nEnter MT5 ticket IDs to investigate, one per line.") print("Type 'done' when finished.\n") tickets = [] while True: t = input("Ticket: ").strip() if t.lower() in ("done", ""): break if t.isdigit(): tickets.append(t) else: print(" not a valid ticket number") for ticket in tickets: analyse_ticket(entries, ticket) if __name__ == "__main__": main()