146 lines
6.0 KiB
Python
146 lines
6.0 KiB
Python
"""Analyze backtest_*.jsonl — print key stats and per-trade attribution."""
|
|
import json
|
|
import sys
|
|
import os
|
|
from collections import defaultdict
|
|
|
|
# Fix Windows console encoding
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def load_jsonl(path):
|
|
entries, exits, snapshots = [], [], []
|
|
with open(path, encoding="utf-8") as f:
|
|
for line in f:
|
|
try:
|
|
e = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
t = e.get("type")
|
|
if t == "entry":
|
|
entries.append(e)
|
|
elif t == "exit":
|
|
exits.append(e)
|
|
elif t == "snapshot":
|
|
snapshots.append(e)
|
|
return entries, exits, snapshots
|
|
|
|
|
|
def stats(entries, exits):
|
|
"""Print overall + segment stats"""
|
|
print(f"\n{'='*70}")
|
|
print(f"OVERALL: {len(entries)} entries, {len(exits)} exits")
|
|
print(f"{'='*70}")
|
|
|
|
wins = [x for x in exits if x["reason"] == "take_profit"]
|
|
losses = [x for x in exits if x["reason"] == "stop_loss"]
|
|
other = [x for x in exits if x["reason"] not in ("take_profit", "stop_loss")]
|
|
|
|
total_pnl = sum(x["pnl"] for x in exits)
|
|
win_pnl = sum(x["pnl"] for x in wins)
|
|
loss_pnl = sum(x["pnl"] for x in losses)
|
|
win_pct = len(wins) / len(exits) * 100 if exits else 0
|
|
|
|
print(f"Wins: {len(wins)} Losses: {len(losses)} Other: {len(other)}")
|
|
print(f"Win rate: {win_pct:.1f}%")
|
|
print(f"Total PnL: ${total_pnl:+.2f}")
|
|
print(f"Avg win: ${win_pnl/len(wins):+.3f}" if wins else "")
|
|
print(f"Avg loss: ${loss_pnl/len(losses):+.3f}" if losses else "")
|
|
print(f"Avg hold (win): {sum(x['hold_seconds'] for x in wins)/len(wins):.1f}s" if wins else "")
|
|
print(f"Avg hold (loss): {sum(x['hold_seconds'] for x in losses)/len(losses):.1f}s" if losses else "")
|
|
|
|
# Pair entries with exits by round_ts
|
|
by_round = defaultdict(dict)
|
|
for e in entries:
|
|
by_round[e["round_ts"]]["entry"] = e
|
|
for x in exits:
|
|
# round_ts lives in state (from capture_market_state at exit)
|
|
rt = x.get("round_ts") or x.get("state", {}).get("round_ts")
|
|
if rt:
|
|
by_round[rt]["exit"] = x
|
|
pairs = [v for v in by_round.values() if "entry" in v and "exit" in v]
|
|
|
|
print(f"\n--- PER TRADE ({len(pairs)} round trips) ---")
|
|
print(f"{'#':>3} {'side':>5} {'conc':>5} {'streak':>9} {'winn':>6} {'price':>5} {'reason':>10} {'pnl':>7} {'hold':>5}")
|
|
for i, p in enumerate(pairs, 1):
|
|
e = p["entry"]
|
|
x = p["exit"]
|
|
s = e["state"]
|
|
streak = f"{e['streak_up']}/{e['streak_dn']}"
|
|
print(f"{i:3d} {e['side'].upper():>5} "
|
|
f"{e['streak_concentration']:+.2f} "
|
|
f"{streak:>9} "
|
|
f"{s['winners_up_n']}/{s['winners_dn_n']:<4} "
|
|
f"{e['price']:.2f} "
|
|
f"{x['reason']:>10} "
|
|
f"${x['pnl']:+.2f} "
|
|
f"{x['hold_seconds']:>4.0f}s")
|
|
|
|
# Segment analysis by streak concentration
|
|
print(f"\n--- BY STREAK CONCENTRATION ---")
|
|
buckets = [(-1.0, -0.7), (-0.7, -0.4), (-0.4, 0.4), (0.4, 0.7), (0.7, 1.01)]
|
|
for lo, hi in buckets:
|
|
seg = [p for p in pairs
|
|
if lo <= p["entry"]["streak_concentration"] < hi]
|
|
if not seg:
|
|
continue
|
|
seg_wins = sum(1 for p in seg if p["exit"]["reason"] == "take_profit")
|
|
seg_pnl = sum(p["exit"]["pnl"] for p in seg)
|
|
avg_pnl = seg_pnl / len(seg)
|
|
print(f" [{lo:+.1f}, {hi:+.1f}) n={len(seg):>3} wins={seg_wins:>3} ({seg_wins/len(seg)*100:.0f}%) PnL=${seg_pnl:+.2f} avg=${avg_pnl:+.3f}")
|
|
|
|
# Segment by entry price (token price tier)
|
|
print(f"\n--- BY ENTRY PRICE ---")
|
|
p_buckets = [(0, 0.4), (0.4, 0.55), (0.55, 0.7), (0.7, 0.85), (0.85, 1.01)]
|
|
for lo, hi in p_buckets:
|
|
seg = [p for p in pairs if lo <= p["entry"]["price"] < hi]
|
|
if not seg:
|
|
continue
|
|
seg_wins = sum(1 for p in seg if p["exit"]["reason"] == "take_profit")
|
|
seg_pnl = sum(p["exit"]["pnl"] for p in seg)
|
|
print(f" [{lo:.2f}, {hi:.2f}) n={len(seg):>3} wins={seg_wins:>3} ({seg_wins/len(seg)*100:.0f}%) PnL=${seg_pnl:+.2f}")
|
|
|
|
# Segment by hold time
|
|
print(f"\n--- BY HOLD SECONDS ---")
|
|
h_buckets = [(0, 30), (30, 60), (60, 120), (120, 300)]
|
|
for lo, hi in h_buckets:
|
|
seg = [p for p in pairs if lo <= p["exit"]["hold_seconds"] < hi]
|
|
if not seg:
|
|
continue
|
|
seg_wins = sum(1 for p in seg if p["exit"]["reason"] == "take_profit")
|
|
seg_pnl = sum(p["exit"]["pnl"] for p in seg)
|
|
print(f" [{lo:>3}s, {hi:>3}s) n={len(seg):>3} wins={seg_wins:>3} ({seg_wins/len(seg)*100:.0f}%) PnL=${seg_pnl:+.2f}")
|
|
|
|
# Per-feature breakdown: book imbalance, BTC velocity at entry
|
|
print(f"\n--- BY BOOK IMBALANCE (positive = UP buy pressure) ---")
|
|
b_buckets = [(-9999, -500), (-500, -100), (-100, 100), (100, 500), (500, 9999)]
|
|
for lo, hi in b_buckets:
|
|
seg = [p for p in pairs if lo <= p["entry"]["state"]["book_imbalance"] < hi]
|
|
if not seg:
|
|
continue
|
|
seg_wins = sum(1 for p in seg if p["exit"]["reason"] == "take_profit")
|
|
seg_pnl = sum(p["exit"]["pnl"] for p in seg)
|
|
print(f" [{lo:>5}, {hi:>5}) n={len(seg):>3} wins={seg_wins:>3} ({seg_wins/len(seg)*100:.0f}%) PnL=${seg_pnl:+.2f}")
|
|
|
|
# BTC deviation at entry
|
|
print(f"\n--- BY BTC DEV AT ENTRY ($) ---")
|
|
d_buckets = [(-999, -10), (-10, -3), (-3, 3), (3, 10), (10, 999)]
|
|
for lo, hi in d_buckets:
|
|
seg = [p for p in pairs if lo <= p["entry"]["state"]["btc_dev"] < hi]
|
|
if not seg:
|
|
continue
|
|
seg_wins = sum(1 for p in seg if p["exit"]["reason"] == "take_profit")
|
|
seg_pnl = sum(p["exit"]["pnl"] for p in seg)
|
|
print(f" [{lo:>4}, {hi:>4}) n={len(seg):>3} wins={seg_wins:>3} ({seg_wins/len(seg)*100:.0f}%) PnL=${seg_pnl:+.2f}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
path = sys.argv[1] if len(sys.argv) > 1 else "backtest_0725_2344.jsonl"
|
|
if not os.path.exists(path):
|
|
print(f"File not found: {path}")
|
|
sys.exit(1)
|
|
entries, exits, snapshots = load_jsonl(path)
|
|
stats(entries, exits) |