""" Trade Log — JSON journal of all trading actions. Records entries, partial closes, SL hits, and calculates PnL. """ import json import os import logging from datetime import datetime, timezone, timedelta logger = logging.getLogger(__name__) TRADE_LOG_DIR = os.path.dirname(os.path.abspath(__file__)) TRADE_LOG_FILE = os.path.join(TRADE_LOG_DIR, "trade_log.json") VANCOUVER_TZ = timezone(timedelta(hours=-7)) def now_van() -> datetime: return datetime.now(VANCOUVER_TZ) def load_trade_log() -> list[dict]: """Load trade log from file.""" if not os.path.exists(TRADE_LOG_FILE): return [] try: with open(TRADE_LOG_FILE, "r") as f: return json.load(f) except Exception as e: logger.error(f"Failed to load trade log: {e}") return [] def save_trade_log(log: list[dict]): """Save trade log to file.""" try: with open(TRADE_LOG_FILE, "w") as f: json.dump(log, f, indent=2, ensure_ascii=False) except Exception as e: logger.error(f"Failed to save trade log: {e}") def log_event(event_type: str, data: dict): """ Append event to trade log. event_type: ENTRY, TP1_HIT, TP2_HIT, TP3_HIT, SL_HIT, MANUAL_CLOSE, ERROR data: symbol, price, quantity, pnl, etc. """ now = now_van() event = { "timestamp": now.isoformat(), "timestamp_utc": datetime.now(timezone.utc).isoformat(), "event": event_type, **data, } log = load_trade_log() log.append(event) save_trade_log(log) logger.info(f"Trade event: {event_type} | {data.get('symbol', '?')} | {data}") return event def get_open_trades() -> list[dict]: """Get entries that don't have a closing event — for recovery on restart.""" log = load_trade_log() # Find symbols with ENTRY but no final close (SL_HIT, TP3_HIT, MANUAL_CLOSE) open_symbols = {} close_events = {"SL_HIT", "TP3_HIT", "MANUAL_CLOSE"} for event in log: symbol = event.get("symbol", "") evt = event.get("event", "") if evt == "ENTRY": open_symbols[symbol] = event elif evt in close_events and symbol in open_symbols: del open_symbols[symbol] return list(open_symbols.values()) def get_tp_state(symbol: str) -> dict: """Get TP state for a symbol from log (which TPs were hit).""" log = load_trade_log() state = {"tp1_hit": False, "tp2_hit": False, "current_sl_pct": None} for event in reversed(log): if event.get("symbol") != symbol: continue evt = event.get("event", "") if evt == "ENTRY": break # Found the entry, stop looking if evt == "TP1_HIT": state["tp1_hit"] = True if evt == "TP2_HIT": state["tp2_hit"] = True if "new_sl_pct" in event: state["current_sl_pct"] = event["new_sl_pct"] return state def _filter_by_period(log: list[dict], period: str) -> list[dict]: """Filter trade log events by time period.""" now = now_van() if period == "today": start = now.replace(hour=0, minute=0, second=0, microsecond=0) elif period == "week": start = now - timedelta(days=7) elif period == "month": start = now - timedelta(days=30) else: return log filtered = [] for e in log: try: ts = datetime.fromisoformat(e["timestamp"]) if ts >= start: filtered.append(e) except Exception: continue return filtered def calculate_pnl_summary(period: str = "all") -> dict: """Calculate PnL summary for WT strategy (non-scalp trades).""" log = load_trade_log() filtered = _filter_by_period(log, period) # Exclude scalp events filtered = [e for e in filtered if not e.get("event", "").startswith("SCALP_")] entries = [e for e in filtered if e.get("event") == "ENTRY"] sl_hits = [e for e in filtered if e.get("event") == "SL_HIT"] tp1_hits = [e for e in filtered if e.get("event") == "TP1_HIT"] tp2_hits = [e for e in filtered if e.get("event") == "TP2_HIT"] tp3_hits = [e for e in filtered if e.get("event") == "TP3_HIT"] manual = [e for e in filtered if e.get("event") == "MANUAL_CLOSE"] total_pnl = sum(e.get("realized_pnl_usdt", 0) for e in filtered if "realized_pnl_usdt" in e) # Win/Loss: WIN = at least TP1 was hit before close # Group by trade_id to determine per-trade outcome trades = {} # trade_id -> {tp1, tp2, tp3, sl, pnl} for e in filtered: tid = e.get("trade_id", "") if not tid: continue if tid not in trades: trades[tid] = {"tp1": False, "tp2": False, "tp3": False, "sl": False, "closed": False} evt = e.get("event", "") if evt == "TP1_HIT": trades[tid]["tp1"] = True elif evt == "TP2_HIT": trades[tid]["tp2"] = True elif evt == "TP3_HIT": trades[tid]["tp3"] = True trades[tid]["closed"] = True elif evt == "SL_HIT": trades[tid]["sl"] = True trades[tid]["closed"] = True elif evt == "MANUAL_CLOSE": trades[tid]["closed"] = True closed_trades = {k: v for k, v in trades.items() if v["closed"]} total_closed = len(closed_trades) wins = sum(1 for v in closed_trades.values() if v["tp1"]) # At least TP1 hit = win losses = total_closed - wins win_rate = (wins / total_closed * 100) if total_closed > 0 else 0 return { "total_entries": len(entries), "total_closed": total_closed, "wins": wins, "losses": losses, "win_rate": round(win_rate, 1), "tp1_hits": len(tp1_hits), "tp2_hits": len(tp2_hits), "tp3_full_hits": len(tp3_hits), "sl_hits": len(sl_hits), "total_pnl_usdt": round(total_pnl, 2), "period": period, } def calculate_scalp_summary(period: str = "all") -> dict: """Calculate PnL summary for Quick Take scalp strategy.""" log = load_trade_log() filtered = _filter_by_period(log, period) # Only scalp events scalp = [e for e in filtered if e.get("event", "").startswith("SCALP_")] entries = [e for e in scalp if e.get("event") == "SCALP_ENTRY"] tp_hits = [e for e in scalp if e.get("event") == "SCALP_TP"] sl_hits = [e for e in scalp if e.get("event") == "SCALP_SL"] time_stops = [e for e in scalp if e.get("event") == "SCALP_TIME_STOP"] manual = [e for e in scalp if e.get("event") == "SCALP_MANUAL"] total_closed = len(tp_hits) + len(sl_hits) + len(time_stops) + len(manual) # Wins = TP + positive time stops + positive manual wins = len(tp_hits) for e in time_stops + manual: if e.get("pnl_pct", 0) >= 0: wins += 1 losses = total_closed - wins win_rate = (wins / total_closed * 100) if total_closed > 0 else 0 total_pnl = sum(e.get("pnl_usdt", 0) for e in scalp if "pnl_usdt" in e) # Average trade duration durations = [e.get("age_minutes", 0) for e in scalp if "age_minutes" in e] avg_duration = sum(durations) / len(durations) if durations else 0 # Best and worst trade pnls = [e.get("pnl_usdt", 0) for e in scalp if "pnl_usdt" in e] best = max(pnls) if pnls else 0 worst = min(pnls) if pnls else 0 return { "total_entries": len(entries), "total_closed": total_closed, "wins": wins, "losses": losses, "win_rate": round(win_rate, 1), "tp_hits": len(tp_hits), "sl_hits": len(sl_hits), "time_stops": len(time_stops), "total_pnl_usdt": round(total_pnl, 4), "avg_duration_min": round(avg_duration, 1), "best_trade": round(best, 4), "worst_trade": round(worst, 4), "period": period, } def calculate_gerchik_summary(period: str = "all") -> dict: """Calculate PnL summary for Gerchik Levels strategy.""" log = load_trade_log() filtered = _filter_by_period(log, period) # Only GR_ events gr = [e for e in filtered if e.get("event", "").startswith("GR_")] entries = [e for e in gr if e.get("event") == "GR_ENTRY"] sl_hits = [e for e in gr if e.get("event") == "GR_SL_HIT"] tp1_hits = [e for e in gr if e.get("event") == "GR_TP1_HIT"] tp3_hits = [e for e in gr if e.get("event") == "GR_TP3_HIT"] manual = [e for e in gr if e.get("event") == "GR_MANUAL_CLOSE"] # Total PnL from close events total_pnl = 0 for e in gr: if "total_trade_pnl_usdt" in e: total_pnl += e["total_trade_pnl_usdt"] elif "total_pnl_usdt" in e: total_pnl += e["total_pnl_usdt"] # Win/Loss: WIN = at least TP1 hit trades = {} for e in gr: tid = e.get("trade_id", "") if not tid: continue if tid not in trades: trades[tid] = {"tp1": False, "closed": False, "model": "?"} evt = e.get("event", "") if evt == "GR_ENTRY": trades[tid]["model"] = e.get("model", "?") elif evt == "GR_TP1_HIT": trades[tid]["tp1"] = True elif evt in ("GR_SL_HIT", "GR_TP3_HIT", "GR_MANUAL_CLOSE"): trades[tid]["closed"] = True closed_trades = {k: v for k, v in trades.items() if v["closed"]} total_closed = len(closed_trades) wins = sum(1 for v in closed_trades.values() if v["tp1"]) losses = total_closed - wins win_rate = (wins / total_closed * 100) if total_closed > 0 else 0 # Count by model model_counts = {"A": 0, "B": 0, "C": 0, "D": 0} for e in entries: m = e.get("model", "?") if m in model_counts: model_counts[m] += 1 return { "total_entries": len(entries), "total_closed": total_closed, "wins": wins, "losses": losses, "win_rate": round(win_rate, 1), "tp1_hits": len(tp1_hits), "tp3_hits": len(tp3_hits), "sl_hits": len(sl_hits), "total_pnl_usdt": round(total_pnl, 2), "model_a": model_counts["A"], "model_b": model_counts["B"], "model_c": model_counts["C"], "model_d": model_counts["D"], "period": period, } def calculate_combined_summary(period: str = "all") -> dict: """Combined stats for all strategies.""" wt = calculate_pnl_summary(period) scalp = calculate_scalp_summary(period) gr = calculate_gerchik_summary(period) return {"wt": wt, "scalp": scalp, "gerchik": gr}