""" compute_perf.py — compute daily P&L summary, send to Telegram, and generate HTML dashboard. Run after market close (e.g., 4:00 PM ET). """ import csv import json import sys from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo import pandas as pd from src.notify import notify ET = ZoneInfo("America/New_York") TRADES_PATH = Path("trades.csv") POSITIONS_PATH = Path("open_positions.json") SAFETY_LOG_PATH = Path("logs/safety-check-log.json") DASHBOARD_DIR = Path("dashboard") def cast_num(v): """Cast numpy/pandas types to Python native types.""" if hasattr(v, "item"): return v.item() return v def compute_daily_perf(): """Compute P&L for closed trades today.""" today_et = datetime.now(ET).date().isoformat() # Load trades if not TRADES_PATH.exists(): trades = [] else: try: trades = list(csv.DictReader(open(TRADES_PATH))) except: trades = [] # Filter to today today_trades = [t for t in trades if t.get("timestamp_iso", "").startswith(today_et)] if not today_trades: return { "success": True, "date": today_et, "total_trades": 0, "wins": 0, "losses": 0, "win_rate_pct": 0.0, "gross_pnl_usd": 0.0, "largest_winner": None, "largest_loser": None, "avg_winner": 0.0, "avg_loser": 0.0, "profit_factor": "n/a", } # Load stop prices from open_positions.json if available stop_prices = {} if POSITIONS_PATH.exists(): try: positions = json.loads(POSITIONS_PATH.read_text()) for pos in positions: stop_prices[pos.get("symbol", "")] = pos.get("initial_stop", 0) except: pass # Pair BUY with SELL by symbol (FIFO) buys = {} closed_pairs = [] for trade in today_trades: symbol = trade.get("symbol", "") side = trade.get("side", "") price = float(trade.get("fill_price", 0)) if trade.get("fill_price") else 0 size = int(trade.get("size", 0)) if trade.get("size") else 0 ts = trade.get("timestamp_iso", "") if side == "BUY": if symbol not in buys: buys[symbol] = [] buys[symbol].append({"price": price, "size": size, "ts": ts}) elif side == "SELL" and symbol in buys and buys[symbol]: buy = buys[symbol].pop(0) if price > 0 and buy["price"] > 0: pnl = (price - buy["price"]) * buy["size"] pnl_pct = ((price - buy["price"]) / buy["price"]) * 100 try: buy_dt = datetime.fromisoformat(buy["ts"]) sell_dt = datetime.fromisoformat(ts) hold_minutes = (sell_dt - buy_dt).total_seconds() / 60 except: hold_minutes = 0 # Calculate per-trade R initial_stop = stop_prices.get(symbol, buy["price"] * 0.99) risk_per_share = buy["price"] - initial_stop R = (price - buy["price"]) / risk_per_share if risk_per_share > 0 else 0 closed_pairs.append({ "symbol": symbol, "buy_price": buy["price"], "sell_price": price, "initial_stop": initial_stop, "size": buy["size"], "pnl": pnl, "pnl_pct": pnl_pct, "hold_minutes": hold_minutes, "R": R, "ts": ts, }) if not closed_pairs: return { "success": True, "date": today_et, "total_trades": 0, "wins": 0, "losses": 0, "win_rate_pct": 0.0, "gross_pnl_usd": 0.0, "largest_winner": None, "largest_loser": None, "avg_winner": 0.0, "avg_loser": 0.0, "profit_factor": "n/a", "closed_pairs": [], } # Aggregate total = len(closed_pairs) pnls = [p["pnl"] for p in closed_pairs] winners = [p for p in closed_pairs if p["pnl"] > 0] losers = [p for p in closed_pairs if p["pnl"] < 0] wins = len(winners) losses = len(losers) win_rate = (wins / total * 100) if total > 0 else 0 gross_pnl = sum(pnls) largest_winner = max(winners, key=lambda x: x["pnl"]) if winners else None largest_loser = min(losers, key=lambda x: x["pnl"]) if losers else None avg_winner = sum(p["pnl"] for p in winners) / len(winners) if winners else 0 avg_loser = sum(p["pnl"] for p in losers) / len(losers) if losers else 0 sum_wins = sum(p["pnl"] for p in winners) if winners else 0 sum_losses = abs(sum(p["pnl"] for p in losers)) if losers else 0 pf = (sum_wins / sum_losses) if sum_losses > 0 else ("inf" if sum_wins > 0 else "n/a") return { "success": True, "date": today_et, "total_trades": total, "wins": wins, "losses": losses, "win_rate_pct": round(win_rate, 2), "gross_pnl_usd": round(gross_pnl, 2), "largest_winner": f"{largest_winner['symbol']} ${largest_winner['pnl']:.2f}" if largest_winner else None, "largest_loser": f"{largest_loser['symbol']} ${largest_loser['pnl']:.2f}" if largest_loser else None, "avg_winner": round(avg_winner, 2), "avg_loser": round(avg_loser, 2), "profit_factor": round(pf, 2) if isinstance(pf, float) else pf, "closed_pairs": closed_pairs, } def get_open_positions(): """Load current open positions.""" if not POSITIONS_PATH.exists(): return [] try: return json.loads(POSITIONS_PATH.read_text()) except: return [] def get_last_cycle_info(): """Get last cycle timestamp from safety-check-log.json.""" if not SAFETY_LOG_PATH.exists(): return None, "No cycle data yet" try: with open(SAFETY_LOG_PATH) as f: lines = f.readlines() if lines: last = json.loads(lines[-1]) ts = last.get("timestamp", "") decision = last.get("decision", "") return ts, decision except: pass return None, "No cycle data yet" def build_r_histogram(closed_pairs): """Build R-multiple histogram buckets.""" buckets = { "(-∞, -2R]": 0, "(-2R, -1R]": 0, "(-1R, 0R]": 0, "(0R, +1R]": 0, "(+1R, +2R]": 0, "(+2R, +3R]": 0, "(+3R, +∞)": 0, } for pair in closed_pairs: R = pair["R"] if R <= -2: buckets["(-∞, -2R]"] += 1 elif R <= -1: buckets["(-2R, -1R]"] += 1 elif R <= 0: buckets["(-1R, 0R]"] += 1 elif R <= 1: buckets["(0R, +1R]"] += 1 elif R <= 2: buckets["(+1R, +2R]"] += 1 elif R <= 3: buckets["(+2R, +3R]"] += 1 else: buckets["(+3R, +∞)"] += 1 return buckets def generate_html_dashboard(perf, open_positions, cycle_ts, cycle_status): """Generate HTML dashboard.""" DASHBOARD_DIR.mkdir(exist_ok=True) closed_pairs = perf.get("closed_pairs", []) r_histogram = build_r_histogram(closed_pairs) max_bucket = max(r_histogram.values()) if r_histogram.values() else 1 # Recent trades (last 20) recent_trades_html = "" if closed_pairs: for pair in closed_pairs[-20:]: r_color = "success" if pair["R"] > 0 else "danger" recent_trades_html += f""" {pair["symbol"]} {pair["size"]} ${pair["buy_price"]:.2f} ${pair["sell_price"]:.2f} ${pair["pnl"]:+.2f} {pair["R"]:.2f}R """ # Open positions table open_pos_html = "" if open_positions: for pos in open_positions: open_pos_html += f""" {pos.get("symbol", "")} {pos.get("qty", "")} ${pos.get("entry_price", 0):.2f} ${pos.get("initial_stop", 0):.2f} {pos.get("state", "")} """ # R histogram bars histogram_html = "" for bucket, count in r_histogram.items(): width_pct = (count / max_bucket * 100) if max_bucket > 0 else 0 histogram_html += f"""
{bucket} {count} trades
""" cycle_time_str = cycle_ts if cycle_ts else "—" bot_status = "ACTIVE" if perf["total_trades"] > 0 else "IDLE" html = f""" IBKR Bot Dashboard

🤖 IBKR Paper-Trading Bot

Status: {bot_status}

Last Cycle: {cycle_time_str}
{perf["date"]}
Today's P&L Summary
{perf["gross_pnl_usd"]:+.2f}
Total P&L
{perf["total_trades"]}
Closed Trades
{perf["wins"]}W / {perf["losses"]}L
Wins / Losses
{perf["win_rate_pct"]}%
Win Rate
R-Multiple Distribution
{histogram_html if closed_pairs else "

No closed trades yet.

"}
Open Positions ({len(open_positions)})
{f''' {open_pos_html if open_positions else ""}
Symbol Qty Entry Stop State
No open positions.
''' if open_positions else "

No open positions.

"}
Recent Closed Trades (Last 20)
{f''' {recent_trades_html if closed_pairs else ""}
Symbol Qty Buy Price Sell Price P&L R Multiple
No closed trades yet.
''' if closed_pairs else "

No closed trades yet.

"}
""" dashboard_file = DASHBOARD_DIR / "index.html" with open(dashboard_file, "w") as f: f.write(html) return dashboard_file def main(): perf = compute_daily_perf() # Preserve existing behavior: JSON stdout perf_out = {k: v for k, v in perf.items() if k != "closed_pairs"} print(json.dumps(perf_out, indent=2)) # Preserve existing behavior: Telegram notification try: if perf["total_trades"] == 0: body = "No closed trades today." else: body = ( f"Trades: {perf['total_trades']} " f"({perf['wins']}W / {perf['losses']}L, {perf['win_rate_pct']}%)\n" f"P&L: ${perf['gross_pnl_usd']:+.2f}\n" f"Best: {perf['largest_winner'] or 'N/A'}\n" f"Worst: {perf['largest_loser'] or 'N/A'}\n" f"PF: {perf['profit_factor']}" ) notify(f"Daily Summary {perf['date']}", body, "default") except: pass # NEW: Generate HTML dashboard try: open_positions = get_open_positions() cycle_ts, cycle_status = get_last_cycle_info() dashboard_file = generate_html_dashboard(perf, open_positions, cycle_ts, cycle_status) print(f"Dashboard saved to {dashboard_file}") except Exception as e: print(f"[WARN] Dashboard generation failed: {e}") if __name__ == "__main__": main()