From ed3bebf28756aa0d8484a933b70c14d0265be447 Mon Sep 17 00:00:00 2001 From: tinkle-community Date: Thu, 23 Jul 2026 10:30:51 +0900 Subject: [PATCH] feat: TPE Bayesian search harness for autopilot risk/throttle params MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replay recorded AI decisions from decision_records under alternative risk-control and throttle parameters, then search the 15-dim space with Optuna multivariate TPE. - extract.py: parse decision cycles, intents (incl. throttled ones), 15m OHLCV and per-cycle prices out of input prompts into CSVs - simulate.py: replay engine mirroring auto_trader_throttle.go semantics (margin-based PnL%% thresholds, intra-candle SL/TP trigger orders, opens/hour + reentry + margin gates); live-config replay reproduces the real account curve (-83%% sim vs -85%% actual) - search.py: TPE search (train/holdout split) + baselines + importances Findings: train-window optima do not survive holdout (overfit); only 2/800 trials are positive across all 3 time folds. Dominant lever is min_confidence (importance 0.79) — the edge problem is decision quality, not risk parameters. --- scripts/optimize/.gitignore | 2 + scripts/optimize/extract.py | 180 +++++++++++++++++++ scripts/optimize/search.py | 137 ++++++++++++++ scripts/optimize/simulate.py | 338 +++++++++++++++++++++++++++++++++++ 4 files changed, 657 insertions(+) create mode 100644 scripts/optimize/.gitignore create mode 100644 scripts/optimize/extract.py create mode 100644 scripts/optimize/search.py create mode 100644 scripts/optimize/simulate.py diff --git a/scripts/optimize/.gitignore b/scripts/optimize/.gitignore new file mode 100644 index 00000000..b3a6b998 --- /dev/null +++ b/scripts/optimize/.gitignore @@ -0,0 +1,2 @@ +.venv/ +data/ diff --git a/scripts/optimize/extract.py b/scripts/optimize/extract.py new file mode 100644 index 00000000..58a60cc9 --- /dev/null +++ b/scripts/optimize/extract.py @@ -0,0 +1,180 @@ +"""Extract a replayable dataset from data/data.db. + +Reads decision_records and emits three CSVs under scripts/optimize/data/: + cycles.csv - one row per decision cycle (timestamp, recorded equity) + decisions.csv - the AI's intended actions per cycle (including throttled ones) + candles.csv - deduplicated per-symbol 15m OHLCV parsed from input prompts + prices.csv - per-cycle current_price for every symbol present in the prompt + +Run: .venv/bin/python extract.py [--db ../../data/data.db] +""" + +import argparse +import csv +import json +import os +import re +import sqlite3 +from datetime import datetime, timedelta, timezone + +SECTION_RE = re.compile(r"^=== (\S+) Market Data ===$") +CURRENT_PRICE_RE = re.compile(r"^current_price = ([\d.eE+-]+)$") +CANDLE_RE = re.compile( + r"^(\d{2})-(\d{2}) (\d{2}):(\d{2})\s+" + r"([\d.eE+-]+)\s+([\d.eE+-]+)\s+([\d.eE+-]+)\s+([\d.eE+-]+)\s+([\d.eE+-]+)\s*$" +) +EQUITY_RE = re.compile(r"Account: Equity ([\d.]+)") +TS_RE = re.compile(r"(\d{4}-\d{2}-\d{2})[ T](\d{2}:\d{2}:\d{2})") + + +def parse_cycle_ts(raw): + m = TS_RE.search(raw) + if not m: + return None + return datetime.strptime( + f"{m.group(1)} {m.group(2)}", "%Y-%m-%d %H:%M:%S" + ).replace(tzinfo=timezone.utc) + + +def candle_ts(cycle_ts, month, day, hour, minute): + """Candle rows carry no year; anchor to the cycle timestamp.""" + ts = datetime(cycle_ts.year, month, day, hour, minute, tzinfo=timezone.utc) + if ts > cycle_ts + timedelta(days=2): + ts = ts.replace(year=cycle_ts.year - 1) + return ts + + +def parse_prompt(prompt, cycle_ts): + equity = None + m = EQUITY_RE.search(prompt) + if m: + equity = float(m.group(1)) + + prices = {} + candles = [] + symbol = None + for line in prompt.splitlines(): + line = line.rstrip() + sm = SECTION_RE.match(line) + if sm: + symbol = sm.group(1).upper() + continue + if symbol is None: + continue + pm = CURRENT_PRICE_RE.match(line) + if pm: + prices[symbol] = float(pm.group(1)) + continue + cm = CANDLE_RE.match(line) + if cm: + ts = candle_ts(cycle_ts, *(int(cm.group(i)) for i in range(1, 5))) + o, h, low, c, v = (float(cm.group(i)) for i in range(5, 10)) + candles.append((symbol, ts, o, h, low, c, v)) + return equity, prices, candles + + +def parse_decisions(raw): + try: + items = json.loads(raw or "[]") + except json.JSONDecodeError: + return [] + out = [] + for d in items if isinstance(items, list) else []: + if not isinstance(d, dict): + continue + out.append( + { + "action": str(d.get("action", "")).strip().lower(), + "symbol": str(d.get("symbol", "")).strip().upper(), + "leverage": d.get("leverage") or 0, + "price": d.get("price") or 0.0, + "stop_loss": d.get("stop_loss") or 0.0, + "take_profit": d.get("take_profit") or 0.0, + "confidence": d.get("confidence") or 0, + "recorded_success": bool(d.get("success")), + "recorded_error": str(d.get("error", "")), + } + ) + return out + + +def main(): + ap = argparse.ArgumentParser() + here = os.path.dirname(os.path.abspath(__file__)) + ap.add_argument("--db", default=os.path.join(here, "..", "..", "data", "data.db")) + ap.add_argument("--out", default=os.path.join(here, "data")) + args = ap.parse_args() + + os.makedirs(args.out, exist_ok=True) + conn = sqlite3.connect(f"file:{args.db}?mode=ro", uri=True) + rows = conn.execute( + "SELECT id, timestamp, input_prompt, decisions FROM decision_records" + " ORDER BY timestamp" + ) + + cycles, decisions = [], [] + candle_map = {} # (symbol, ts) -> row, latest observation wins + price_rows = [] + skipped = 0 + + for rec_id, ts_raw, prompt, decisions_raw in rows: + cycle_ts = parse_cycle_ts(str(ts_raw)) + if cycle_ts is None: + skipped += 1 + continue + equity, prices, candles = parse_prompt(prompt or "", cycle_ts) + cycles.append((rec_id, cycle_ts.isoformat(), equity if equity is not None else "")) + for sym, price in prices.items(): + price_rows.append((rec_id, cycle_ts.isoformat(), sym, price)) + for sym, cts, o, h, low, c, v in candles: + candle_map[(sym, cts)] = (sym, cts.isoformat(), o, h, low, c, v) + for d in parse_decisions(decisions_raw): + decisions.append( + ( + rec_id, + cycle_ts.isoformat(), + d["action"], + d["symbol"], + d["leverage"], + d["price"], + d["stop_loss"], + d["take_profit"], + d["confidence"], + int(d["recorded_success"]), + d["recorded_error"], + ) + ) + + def write(name, header, data): + path = os.path.join(args.out, name) + with open(path, "w", newline="") as f: + w = csv.writer(f) + w.writerow(header) + w.writerows(data) + return path + + write("cycles.csv", ["cycle_id", "ts", "equity"], cycles) + write( + "decisions.csv", + [ + "cycle_id", "ts", "action", "symbol", "leverage", "price", + "stop_loss", "take_profit", "confidence", "recorded_success", + "recorded_error", + ], + decisions, + ) + write( + "candles.csv", + ["symbol", "ts", "open", "high", "low", "close", "volume"], + sorted(candle_map.values(), key=lambda r: (r[0], r[1])), + ) + write("prices.csv", ["cycle_id", "ts", "symbol", "price"], price_rows) + + print( + f"cycles={len(cycles)} decisions={len(decisions)}" + f" candles={len(candle_map)} prices={len(price_rows)} skipped={skipped}" + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/optimize/search.py b/scripts/optimize/search.py new file mode 100644 index 00000000..f137e7c7 --- /dev/null +++ b/scripts/optimize/search.py @@ -0,0 +1,137 @@ +"""Multivariate TPE Bayesian search over NOFX autopilot risk/throttle params. + +Optimizes on a train window (first ~70% of history) and reports the untouched +holdout window, plus full-period metrics and baselines for reference. + +Run: .venv/bin/python search.py [--trials 800] +""" + +import argparse +import json +import os +from datetime import timedelta + +import optuna + +from simulate import Params, Simulator, load_dataset + +HERE = os.path.dirname(os.path.abspath(__file__)) + + +def suggest_params(trial): + min_hold = trial.suggest_float("min_hold_h", 0.0, 8.0) + return Params( + min_confidence=trial.suggest_int("min_confidence", 70, 95), + min_hold_h=min_hold, + noise_hold_extra_h=trial.suggest_float("noise_hold_extra_h", 0.0, 12.0), + reentry_h=trial.suggest_float("reentry_h", 0.0, 6.0), + max_opens_per_hour=trial.suggest_int("max_opens_per_hour", 1, 6), + max_opens_per_cycle=trial.suggest_int("max_opens_per_cycle", 1, 3), + max_positions=trial.suggest_int("max_positions", 1, 4), + leverage=trial.suggest_int("leverage", 3, 20), + ratio=trial.suggest_float("ratio", 1.0, 6.0), + sl_bypass=trial.suggest_float("sl_bypass", -60.0, -5.0), + tp_bypass=trial.suggest_float("tp_bypass", 5.0, 60.0), + noise_floor=trial.suggest_float("noise_floor", -30.0, -2.0), + noise_ceiling=trial.suggest_float("noise_ceiling", 2.0, 30.0), + sl_mult=trial.suggest_float("sl_mult", 0.5, 2.5), + tp_mult=trial.suggest_float("tp_mult", 0.5, 2.5), + ) + + +def score(metrics): + if metrics is None: + return -1000.0 + if metrics["bankrupt"]: + return -1000.0 + return metrics["ret_pct"] - 0.5 * metrics["max_dd_pct"] + + +def fmt(metrics): + if metrics is None: + return "n/a" + return ( + f"ret {metrics['ret_pct']:+7.1f}% | dd {metrics['max_dd_pct']:5.1f}%" + f" | sharpe {metrics['sharpe']:+5.2f} | trades {metrics['trades']:4d}" + f" | win {metrics['win_rate']:4.1f}% | fees ${metrics['fees']:.0f}" + f" | liq {metrics['liquidations']}" + ) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--trials", type=int, default=800) + ap.add_argument("--seed", type=int, default=42) + args = ap.parse_args() + + dataset = load_dataset() + sim = Simulator(dataset) + cycles = dataset[0] + t0, t1 = cycles[0][1], cycles[-1][1] + split = t0 + (t1 - t0) * 0.7 + print(f"history {t0:%Y-%m-%d} .. {t1:%Y-%m-%d}, holdout from {split:%Y-%m-%d}") + + def objective(trial): + p = suggest_params(trial) + return score(sim.run(p, end=split)) + + sampler = optuna.samplers.TPESampler( + multivariate=True, group=True, seed=args.seed, n_startup_trials=60 + ) + optuna.logging.set_verbosity(optuna.logging.WARNING) + study = optuna.create_study(direction="maximize", sampler=sampler) + study.optimize(objective, n_trials=args.trials, show_progress_bar=True) + + best = Params(**study.best_params) + baselines = { + "live config (2x5x @10x)": Params(), + "old aggressive (4x5x @20x)": Params( + leverage=20, ratio=5.0, max_positions=4, min_hold_h=1.0, + noise_hold_extra_h=0.5, reentry_h=0.5, max_opens_per_hour=6, + max_opens_per_cycle=3, sl_bypass=-2.5, tp_bypass=5.0, + noise_floor=-1.0, noise_ceiling=2.0, + ), + } + + print("\n=== best params (train objective" + f" {study.best_value:+.1f}) ===") + for k, v in sorted(study.best_params.items()): + print(f" {k:22s} = {v:.2f}" if isinstance(v, float) else + f" {k:22s} = {v}") + + print("\n=== best params performance ===") + print(" train :", fmt(sim.run(best, end=split))) + print(" holdout:", fmt(sim.run(best, start=split))) + print(" full :", fmt(sim.run(best))) + + print("\n=== baselines ===") + for name, p in baselines.items(): + print(f" {name}") + print(" train :", fmt(sim.run(p, end=split))) + print(" holdout:", fmt(sim.run(p, start=split))) + + top = sorted( + (t for t in study.trials if t.value is not None), + key=lambda t: t.value, reverse=True + )[:10] + print("\n=== top-10 trials: holdout robustness ===") + for t in top: + m = sim.run(Params(**t.params), start=split) + print(f" train {t.value:+7.1f} | holdout {fmt(m)}") + + try: + imp = optuna.importance.get_param_importances(study) + print("\n=== param importances ===") + for k, v in imp.items(): + print(f" {k:22s} {v:.3f}") + except Exception as e: # sklearn not installed etc. + print(f"\n(param importances unavailable: {e})") + + out = os.path.join(HERE, "data", "best_params.json") + with open(out, "w") as f: + json.dump(study.best_params, f, indent=2) + print(f"\nbest params saved to {out}") + + +if __name__ == "__main__": + main() diff --git a/scripts/optimize/simulate.py b/scripts/optimize/simulate.py new file mode 100644 index 00000000..347c1f12 --- /dev/null +++ b/scripts/optimize/simulate.py @@ -0,0 +1,338 @@ +"""Replay recorded AI decisions under alternative risk/throttle parameters. + +Mirrors the live semantics of trader/auto_trader_throttle.go: + - throttle thresholds compare MARGIN-based PnL% (leverage-multiplied price move), + because auto_trader_loop.go computes UnrealizedPnLPct = pnl / margin + - stop-loss / take-profit are exchange trigger orders -> intra-candle fills + - opens are gated by confidence, per-cycle/per-hour caps, re-entry cooldown, + max positions and available margin + +Known limitation: decisions are replayed as recorded. Parameters that would have +changed WHAT the AI decided (prompt wording, candidate pool) are out of scope; +parameters that gate/size/exit those decisions are faithfully simulated. +""" + +import bisect +import csv +import os +from dataclasses import dataclass, field +from datetime import datetime, timedelta + +HERE = os.path.dirname(os.path.abspath(__file__)) +DATA = os.path.join(HERE, "data") + +FEE_RATE = 0.000784 # measured from trader_fills: 7.84 bps per side +MIN_POSITION_USD = 12.0 +LIQUIDATION_MARGIN_PNL = -0.90 # liquidate when margin PnL <= -90% +DATA_GAP_FORCE_CLOSE = timedelta(hours=48) + + +@dataclass(frozen=True) +class Params: + min_confidence: float = 78.0 + min_hold_h: float = 4.0 + noise_hold_extra_h: float = 4.0 # noise window = min_hold + extra + reentry_h: float = 3.0 + max_opens_per_hour: int = 3 + max_opens_per_cycle: int = 2 + max_positions: int = 2 + leverage: float = 10.0 + ratio: float = 5.0 # per-position notional = ratio x equity + sl_bypass: float = -5.0 # margin-PnL% allowing early AI close + tp_bypass: float = 12.0 + noise_floor: float = -4.0 # margin-PnL% band blocking flat closes + noise_ceiling: float = 6.0 + sl_mult: float = 1.0 # scale AI stop distance from entry + tp_mult: float = 1.0 + margin_cap: float = 1.0 + + +@dataclass +class Position: + symbol: str + side: str # "long" | "short" + entry: float + notional: float # USD at entry + qty: float + margin: float + entry_ts: datetime + stop: float = 0.0 + take: float = 0.0 + last_price: float = 0.0 + last_seen: datetime = None + + +@dataclass +class Trade: + symbol: str + side: str + entry_ts: datetime + exit_ts: datetime + entry: float + exit: float + notional: float + pnl: float # net of fees + fees: float + reason: str + + +def _ts(s): + return datetime.fromisoformat(s) + + +def load_dataset(): + cycles = [] + with open(os.path.join(DATA, "cycles.csv")) as f: + for r in csv.DictReader(f): + cycles.append( + (int(r["cycle_id"]), _ts(r["ts"]), + float(r["equity"]) if r["equity"] else None) + ) + cycles.sort(key=lambda c: c[1]) + + decisions = {} + with open(os.path.join(DATA, "decisions.csv")) as f: + for r in csv.DictReader(f): + decisions.setdefault(int(r["cycle_id"]), []).append( + { + "action": r["action"], + "symbol": r["symbol"], + "price": float(r["price"]), + "stop_loss": float(r["stop_loss"]), + "take_profit": float(r["take_profit"]), + "confidence": float(r["confidence"]), + } + ) + + prices = {} + with open(os.path.join(DATA, "prices.csv")) as f: + for r in csv.DictReader(f): + prices[(int(r["cycle_id"]), r["symbol"])] = float(r["price"]) + + candles = {} + with open(os.path.join(DATA, "candles.csv")) as f: + for r in csv.DictReader(f): + candles.setdefault(r["symbol"], []).append( + (_ts(r["ts"]), float(r["open"]), float(r["high"]), + float(r["low"]), float(r["close"])) + ) + for sym in candles: + candles[sym].sort(key=lambda c: c[0]) + candle_times = {s: [c[0] for c in rows] for s, rows in candles.items()} + return cycles, decisions, prices, candles, candle_times + + +class Simulator: + def __init__(self, dataset, start_equity=None): + self.cycles, self.decisions, self.prices, self.candles, self.candle_times = dataset + recorded = next((e for _, _, e in self.cycles if e), 100.0) + self.start_equity = start_equity if start_equity is not None else recorded + + def price_at(self, symbol, cycle_id, ts): + p = self.prices.get((cycle_id, symbol)) + if p: + return p + times = self.candle_times.get(symbol) + if not times: + return None + i = bisect.bisect_right(times, ts) - 1 + return self.candles[symbol][i][4] if i >= 0 else None + + def margin_pnl_pct(self, pos, price, leverage): + move = (price - pos.entry) / pos.entry + if pos.side == "short": + move = -move + return move * leverage * 100.0 + + def run(self, p: Params, start=None, end=None): + equity = self.start_equity + positions = {} + open_times = [] # for the opens-per-hour cap + last_close = {} # symbol -> ts + trades = [] + curve = [] + min_hold = timedelta(hours=p.min_hold_h) + noise_hold = timedelta(hours=p.min_hold_h + p.noise_hold_extra_h) + reentry = timedelta(hours=p.reentry_h) + + cycles = [c for c in self.cycles + if (start is None or c[1] >= start) and (end is None or c[1] < end)] + if not cycles: + return None + + def close_position(pos, price, ts, reason): + nonlocal equity + move = (price - pos.entry) / pos.entry + if pos.side == "short": + move = -move + gross = pos.notional * move + fees = (pos.notional + pos.qty * price) * FEE_RATE + equity += gross - fees + trades.append(Trade(pos.symbol, pos.side, pos.entry_ts, ts, + pos.entry, price, pos.notional, gross - fees, + fees, reason)) + del positions[pos.symbol] + last_close[pos.symbol] = ts + + for idx, (cycle_id, ts, _) in enumerate(cycles): + next_ts = cycles[idx + 1][1] if idx + 1 < len(cycles) else ts + + # 1. Candle window since previous cycle: trigger orders + liquidation + prev_ts = cycles[idx - 1][1] if idx > 0 else ts - timedelta(minutes=30) + for pos in list(positions.values()): + times = self.candle_times.get(pos.symbol, []) + lo = bisect.bisect_right(times, prev_ts) + hi = bisect.bisect_right(times, ts) + for cts, o, h, low, c in self.candles.get(pos.symbol, [])[lo:hi]: + pos.last_price, pos.last_seen = c, cts + liq_move = -(1.0 / p.leverage) * (-LIQUIDATION_MARGIN_PNL) + if pos.side == "long": + liq_px = pos.entry * (1 + liq_move) + if low <= liq_px: + close_position(pos, liq_px, cts, "liquidation"); break + if pos.stop and low <= pos.stop: + close_position(pos, pos.stop, cts, "stop_loss"); break + if pos.take and h >= pos.take: + close_position(pos, pos.take, cts, "take_profit"); break + else: + liq_px = pos.entry * (1 - liq_move) + if h >= liq_px: + close_position(pos, liq_px, cts, "liquidation"); break + if pos.stop and h >= pos.stop: + close_position(pos, pos.stop, cts, "stop_loss"); break + if pos.take and low <= pos.take: + close_position(pos, pos.take, cts, "take_profit"); break + + # Stale market data: force-close what we can no longer price + for pos in list(positions.values()): + if pos.last_seen and ts - pos.last_seen > DATA_GAP_FORCE_CLOSE: + close_position(pos, pos.last_price, ts, "data_gap") + + if equity <= 0: + return self._metrics(equity, trades, curve, bankrupt=True) + + # 2. Replay this cycle's AI intents + opens_this_cycle = 0 + for d in self.decisions.get(cycle_id, []): + sym, act = d["symbol"], d["action"] + if act in ("close_long", "close_short"): + pos = positions.get(sym) + side = "long" if act == "close_long" else "short" + if not pos or pos.side != side: + continue + price = self.price_at(sym, cycle_id, ts) or pos.last_price + if not price: + continue + pnl_pct = self.margin_pnl_pct(pos, price, p.leverage) + held = ts - pos.entry_ts + if held >= min_hold: + allowed = (held >= noise_hold + or pnl_pct <= p.noise_floor + or pnl_pct >= p.noise_ceiling) + else: + allowed = pnl_pct <= p.sl_bypass or pnl_pct >= p.tp_bypass + if allowed: + close_position(pos, price, ts, "ai_close") + + elif act in ("open_long", "open_short"): + if d["confidence"] < p.min_confidence: + continue + if opens_this_cycle >= p.max_opens_per_cycle: + continue + if sym in positions or len(positions) >= p.max_positions: + continue + hour_ago = ts - timedelta(hours=1) + open_times[:] = [t for t in open_times if t >= hour_ago] + if len(open_times) >= p.max_opens_per_hour: + continue + if sym in last_close and ts - last_close[sym] < reentry: + continue + price = self.price_at(sym, cycle_id, ts) or d["price"] + if not price or price <= 0: + continue + notional = p.ratio * equity + margin_used = sum(x.margin for x in positions.values()) + margin_free = p.margin_cap * equity - margin_used + notional = min(notional, max(0.0, margin_free) * p.leverage) + if notional < MIN_POSITION_USD: + continue + side = "long" if act == "open_long" else "short" + stop = take = 0.0 + if d["stop_loss"] > 0: + stop = price - (price - d["stop_loss"]) * p.sl_mult \ + if side == "long" else \ + price + (d["stop_loss"] - price) * p.sl_mult + if (side == "long") != (stop < price): + stop = 0.0 + if d["take_profit"] > 0: + take = price + (d["take_profit"] - price) * p.tp_mult \ + if side == "long" else \ + price - (price - d["take_profit"]) * p.tp_mult + if (side == "long") != (take > price): + take = 0.0 + equity -= notional * FEE_RATE + positions[sym] = Position( + symbol=sym, side=side, entry=price, notional=notional, + qty=notional / price, margin=notional / p.leverage, + entry_ts=ts, stop=stop, take=take, + last_price=price, last_seen=ts, + ) + open_times.append(ts) + opens_this_cycle += 1 + + # 3. Mark to market + mtm = equity + for pos in positions.values(): + price = self.price_at(pos.symbol, cycle_id, ts) or pos.last_price + move = (price - pos.entry) / pos.entry + if pos.side == "short": + move = -move + mtm += pos.notional * move + curve.append((ts, mtm)) + if mtm <= 0: + return self._metrics(equity, trades, curve, bankrupt=True) + + for pos in list(positions.values()): + close_position(pos, pos.last_price or pos.entry, + cycles[-1][1], "end_of_data") + curve.append((cycles[-1][1], equity)) + return self._metrics(equity, trades, curve, bankrupt=False) + + def _metrics(self, equity, trades, curve, bankrupt): + peak, max_dd = -1e18, 0.0 + for _, v in curve: + peak = max(peak, v) + if peak > 0: + max_dd = max(max_dd, (peak - v) / peak) + daily = {} + for ts, v in curve: + daily[ts.date()] = v + vals = list(daily.values()) + rets = [(b - a) / a for a, b in zip(vals, vals[1:]) if a > 0] + sharpe = 0.0 + if len(rets) > 1: + mean = sum(rets) / len(rets) + var = sum((r - mean) ** 2 for r in rets) / (len(rets) - 1) + if var > 0: + sharpe = mean / var ** 0.5 * (365 ** 0.5) + wins = sum(1 for t in trades if t.pnl > 0) + return { + "final_equity": equity, + "net_pnl": equity - self.start_equity, + "ret_pct": (equity / self.start_equity - 1) * 100, + "max_dd_pct": max_dd * 100, + "sharpe": sharpe, + "trades": len(trades), + "win_rate": wins / len(trades) * 100 if trades else 0.0, + "fees": sum(t.fees for t in trades), + "liquidations": sum(1 for t in trades if t.reason == "liquidation"), + "bankrupt": bankrupt, + } + + +if __name__ == "__main__": + sim = Simulator(load_dataset()) + live = Params() + m = sim.run(live) + print("live-config replay:", {k: round(v, 2) if isinstance(v, float) else v + for k, v in m.items()})