""" Penta-Core Quantitative Evaluation (BTC, ETH, SOL, AVAX, ADA) ============================================================= Evaluates the 5-asset universe without DOGE across: 1. 2-Year Regime (2024-09-23 to 2026-09-23) 2. 5-Year Full Regime (2021-10-26 to 2026-09-23) """ import os import json import numpy as np import pandas as pd from datetime import datetime from pattern_detector import PatternDetector BASE_DIR = r"C:\JevTrader" DATA_DIR = os.path.join(BASE_DIR, "historical_data") SYMBOLS = ["BTC", "ETH", "SOL", "AVAX", "ADA"] ASSET_NAMES = { "BTC": "Bitcoin (BTC/USD)", "ETH": "Ethereum (ETH/USD)", "SOL": "Solana (SOL/USD)", "AVAX": "Avalanche (AVAX/USD)", "ADA": "Cardano (ADA/USD)" } print("1. Loading historical 4H continuous data for Penta-Core...") dfs = {} for sym in SYMBOLS: p = os.path.join(DATA_DIR, f"{sym}_USD_4h_5yr.csv") df = pd.read_csv(p) df.columns = [c.lower() for c in df.columns] df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True).dt.tz_localize(None) dfs[sym] = df ts_sets = [set(df['timestamp']) for df in dfs.values()] common_ts = sorted(list(set.intersection(*ts_sets))) print(f"Penta-Core aligned: {len(common_ts)} timestamps across {SYMBOLS}") aligned = {sym: df[df['timestamp'].isin(common_ts)].sort_values('timestamp').reset_index(drop=True) for sym, df in dfs.items()} detector = PatternDetector(pivot_window=4, tolerance_pct=0.025) processed = {} flags = {} for sym, df in aligned.items(): d = df.copy() d['ema_200'] = d['close'].ewm(span=200, adjust=False).mean() d['upper_14'] = d['high'].shift(1).rolling(14).max() d['lower_14'] = d['low'].shift(1).rolling(14).min() d['vol_sma'] = d['volume'].rolling(20).mean() d['mom_14'] = (d['close'] - d['close'].shift(14)) / d['close'].shift(14) tr1 = d['high'] - d['low'] tr2 = (d['high'] - d['close'].shift(1)).abs() tr3 = (d['low'] - d['close'].shift(1)).abs() d['tr'] = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) d['atr_14'] = d['tr'].rolling(14).mean() fdict = {} for p in detector.scan_all_patterns(d): fdict.setdefault(p['confirm_idx'], []).append(p) processed[sym] = d flags[sym] = fdict def run_sim(start_date, initial_capital=24640.0, staged=False, warmup_bars=70): start_idx = [i for i, t in enumerate(common_ts) if t >= start_date][0] sub_ts = common_ts[start_idx:] sub_data = {sym: df.iloc[start_idx:].reset_index(drop=True) for sym, df in processed.items()} sub_flags = {sym: {idx - start_idx: plist for idx, plist in fdict.items() if idx >= start_idx} for sym, fdict in flags.items()} n_bars = len(sub_ts) current_equity = initial_capital peak_equity = initial_capital trail_atr = 3.4 long_risk_pct = 0.020 short_risk_pct = 0.008 fee_pct = 0.0006 slip_pct = 0.0003 vol_surge_mult = 1.75 stale_exit_bars = 8 max_basket_leverage = 2.5 single_asset_max_lev = 0.75 max_concurrent = 3 m6_idx = n_bars // (10 if len(sub_ts) > 5000 else 4) m12_idx = (n_bars * 2) // (10 if len(sub_ts) > 5000 else 4) m18_idx = (n_bars * 3) // (10 if len(sub_ts) > 5000 else 4) m6_inj, m12_inj, m18_inj = False, False, False total_injected = initial_capital active_pos = {} trades_ledger = [] trade_id = 0 total_fees = 0.0 equity_curve = [] cb_triggers = 0 cb_in_effect = False cb_active_bars = 0 max_dd = 0.0 for i in range(warmup_bars, n_bars): if staged: if i >= m6_idx and not m6_inj: current_equity += 48000.0; total_injected += 48000.0; m6_inj = True if i >= m12_idx and not m12_inj: current_equity += 48000.0; total_injected += 48000.0; m12_inj = True if i >= m18_idx and not m18_inj: current_equity += 50400.0; total_injected += 50400.0; m18_inj = True t = sub_ts[i] peak_equity = max(peak_equity, current_equity) current_dd = (current_equity - peak_equity) / peak_equity if peak_equity > 0 else 0.0 max_dd = min(max_dd, current_dd) circuit_breaker_active = current_dd <= -0.20 if circuit_breaker_active: cb_active_bars += 1 if not cb_in_effect: cb_triggers += 1 cb_in_effect = True else: cb_in_effect = False # Manage active positions for sym in list(active_pos.keys()): pos = active_pos[sym] row = sub_data[sym].iloc[i] c, h, l, a = float(row['close']), float(row['high']), float(row['low']), float(row['atr_14']) entry_p = pos['entry_price'] sl = pos['stop_loss'] alloc = pos['allocated_usd'] sl_dist = pos['sl_dist'] bars_held = i - pos['entry_idx'] exit_trig = False exit_price = 0.0 if pos['direction'] == "LONG": if not pos['tier2'] and c >= entry_p + (1.5 * sl_dist): add = min(current_equity * 0.15, max(0.0, current_equity * 0.90 - pos['allocated_usd'])) if add > 50.0: pos['allocated_usd'] += add alloc = pos['allocated_usd'] pos['tier2'] = True pos['stop_loss'] = max(pos['stop_loss'], entry_p + (0.10 * sl_dist)) sl = pos['stop_loss'] if not pos['tier3'] and c >= entry_p + (3.0 * sl_dist): add = min(current_equity * 0.10, max(0.0, current_equity * 1.00 - pos['allocated_usd'])) if add > 50.0: pos['allocated_usd'] += add alloc = pos['allocated_usd'] pos['tier3'] = True pos['stop_loss'] = max(pos['stop_loss'], entry_p + (1.50 * sl_dist)) sl = pos['stop_loss'] new_sl = max(sl, h - (a * trail_atr)) pos['stop_loss'] = new_sl if bars_held >= stale_exit_bars and c <= entry_p: exit_trig = True; exit_price = c * (1.0 - slip_pct); reason = "32h Stale Decay" elif l <= pos['stop_loss']: exit_trig = True; exit_price = pos['stop_loss'] * (1.0 - slip_pct); reason = "Trailing ATR Stop" if exit_trig: trade_id += 1 ret_pct = (exit_price - entry_p) / entry_p pnl_usd = alloc * ret_pct fee = alloc * (1.0 + ret_pct) * fee_pct + pos['entry_fee'] total_fees += fee net_pnl = pnl_usd - fee current_equity += net_pnl tier = "Tier 3 (+3.0R)" if pos['tier3'] else ("Tier 2 (+1.5R)" if pos['tier2'] else "Tier 1 Base") trades_ledger.append({ "id": trade_id, "symbol": sym, "asset_name": ASSET_NAMES[sym], "direction": "LONG", "entry_date": pos['entry_time'].strftime('%Y-%m-%d %H:%M'), "exit_date": t.strftime('%Y-%m-%d %H:%M'), "entry_ts": int(pos['entry_time'].timestamp()), "exit_ts": int(t.timestamp()), "hold_days": round(bars_held * 4.0 / 24.0, 2), "entry_price": round(entry_p, 4 if entry_p < 10 else 2), "exit_price": round(exit_price, 4 if exit_price < 10 else 2), "pos_size": round(alloc, 2), "gross_ret": f"{ret_pct * 100:+.2f}%", "net_pnl": round(net_pnl, 2), "fee": round(fee, 2), "tier_level": tier, "exit_reason": reason, "is_win": net_pnl > 0 }) del active_pos[sym] else: # SHORT new_sl = min(sl, l + (a * trail_atr)) pos['stop_loss'] = new_sl if bars_held >= stale_exit_bars and c >= entry_p: exit_trig = True; exit_price = c * (1.0 + slip_pct); reason = "32h Stale Decay" elif h >= pos['stop_loss']: exit_trig = True; exit_price = pos['stop_loss'] * (1.0 + slip_pct); reason = "Trailing ATR Stop" if exit_trig: trade_id += 1 ret_pct = (entry_p - exit_price) / entry_p pnl_usd = alloc * ret_pct fee = alloc * (1.0 + ret_pct) * fee_pct + pos['entry_fee'] total_fees += fee net_pnl = pnl_usd - fee current_equity += net_pnl trades_ledger.append({ "id": trade_id, "symbol": sym, "asset_name": ASSET_NAMES[sym], "direction": "SHORT", "entry_date": pos['entry_time'].strftime('%Y-%m-%d %H:%M'), "exit_date": t.strftime('%Y-%m-%d %H:%M'), "entry_ts": int(pos['entry_time'].timestamp()), "exit_ts": int(t.timestamp()), "hold_days": round(bars_held * 4.0 / 24.0, 2), "entry_price": round(entry_p, 4 if entry_p < 10 else 2), "exit_price": round(exit_price, 4 if exit_price < 10 else 2), "pos_size": round(alloc, 2), "gross_ret": f"{ret_pct * 100:+.2f}%", "net_pnl": round(net_pnl, 2), "fee": round(fee, 2), "tier_level": "Tier 1 Base", "exit_reason": reason, "is_win": net_pnl > 0 }) del active_pos[sym] # Entries total_alloc = sum(p['allocated_usd'] for p in active_pos.values()) avail_basket_cap = max(0.0, (current_equity * max_basket_leverage) - total_alloc) if len(active_pos) < max_concurrent and not circuit_breaker_active and avail_basket_cap > 50.0: rankings = [] for s in SYMBOLS: if s not in active_pos: mom = float(sub_data[s].iloc[i]['mom_14']) rankings.append((s, mom)) rankings.sort(key=lambda x: x[1], reverse=True) for rank_idx, (s, _) in enumerate(rankings): if len(active_pos) >= max_concurrent: break tilt = 1.35 if rank_idx == 0 else (1.00 if rank_idx == 1 else 0.70) df_ref = sub_data[s] fdict = sub_flags[s] r = df_ref.iloc[i] c, o, h, l, v = float(r['close']), float(r['open']), float(r['high']), float(r['low']), float(r['volume']) ema, u14, l14, a, vsma = float(r['ema_200']), float(r['upper_14']), float(r['lower_14']), float(r['atr_14']), float(r['vol_sma']) rng = max(h - l, 0.0001) u_wick = (h - max(o, c)) / rng l_wick = (min(o, c) - l) / rng has_bull = any(p["direction"] == "LONG" for off in [0, 1, 2] if (i - off) in fdict for p in fdict[i - off]) has_bear = any(p["direction"] == "SHORT" for off in [0, 1, 2] if (i - off) in fdict for p in fdict[i - off]) long_trig = (c > u14 and c > ema and has_bull and u_wick <= 0.35 and v >= vol_surge_mult * vsma) short_trig = (c < l14 and c < ema and has_bear and l_wick <= 0.35 and v >= 0.50 * vsma) if long_trig: sl_dist = a * trail_atr risk_usd = current_equity * long_risk_pct * tilt pos_size = risk_usd / (sl_dist / c) if sl_dist > 0 else 0.0 pos_size = min(pos_size, current_equity * single_asset_max_lev, avail_basket_cap) if pos_size >= 50.0: entry_p = c * (1.0 + slip_pct) efee = pos_size * fee_pct active_pos[s] = { "direction": "LONG", "entry_price": entry_p, "entry_time": t, "entry_idx": i, "allocated_usd": pos_size, "stop_loss": entry_p - sl_dist, "sl_dist": sl_dist, "tier2": False, "tier3": False, "entry_fee": efee } total_fees += efee avail_basket_cap -= pos_size elif short_trig: sl_dist = a * trail_atr risk_usd = current_equity * short_risk_pct * tilt pos_size = risk_usd / (sl_dist / c) if sl_dist > 0 else 0.0 pos_size = min(pos_size, current_equity * single_asset_max_lev, avail_basket_cap) if pos_size >= 50.0: entry_p = c * (1.0 - slip_pct) efee = pos_size * fee_pct active_pos[s] = { "direction": "SHORT", "entry_price": entry_p, "entry_time": t, "entry_idx": i, "allocated_usd": pos_size, "stop_loss": entry_p + sl_dist, "sl_dist": sl_dist, "tier2": False, "tier3": False, "entry_fee": efee } total_fees += efee avail_basket_cap -= pos_size equity_curve.append({"time": int(t.timestamp()), "value": round(current_equity, 2)}) wins = [t for t in trades_ledger if t['is_win']] losses = [t for t in trades_ledger if not t['is_win']] gw = sum(t['net_pnl'] for t in wins) gl = abs(sum(t['net_pnl'] for t in losses)) pf = gw / gl if gl > 0 else 0.0 years = (sub_ts[-1] - sub_ts[0]).days / 365.25 cagr = ((current_equity / total_injected) ** (1.0 / years)) - 1.0 summary = { "initial_capital": round(initial_capital, 2), "total_injected": round(total_injected, 2), "final_equity": round(current_equity, 2), "net_profit": round(current_equity - total_injected, 2), "return_pct": round((current_equity - total_injected) / total_injected * 100, 2), "cagr": round(cagr * 100, 2), "max_dd": round(abs(max_dd) * 100, 2), "calmar": round(cagr / abs(max_dd), 2) if max_dd != 0 else 0.0, "pf": round(pf, 2), "win_rate": round(len(wins) / len(trades_ledger) * 100, 1) if trades_ledger else 0.0, "total_trades": len(trades_ledger), "wins_count": len(wins), "losses_count": len(losses), "payoff_ratio": round((gw / len(wins)) / (gl / len(losses)), 2) if wins and losses else 0.0, "avg_win": round(gw / len(wins), 2) if wins else 0.0, "avg_loss": round(gl / len(losses), 2) if losses else 0.0, "total_fees": round(total_fees, 2), "cb_triggers": cb_triggers, "cb_active_bars": cb_active_bars, "years": round(years, 2) } return summary, equity_curve, trades_ledger # 2-Year Run start_2yr = pd.to_datetime("2024-09-23 00:00:00") sum_2yr, eq_2yr, trades_2yr = run_sim(start_2yr, initial_capital=24640.0, staged=False, warmup_bars=70) # 5-Year Staged Run start_5yr = pd.to_datetime("2021-09-23 00:00:00") sum_5yr_staged, eq_5yr_staged, trades_5yr_staged = run_sim(start_5yr, initial_capital=24640.0, staged=True, warmup_bars=200) # 5-Year Seed Run sum_5yr_seed, eq_5yr_seed, trades_5yr_seed = run_sim(start_5yr, initial_capital=24640.0, staged=False, warmup_bars=200) print("=" * 80) print("PENTA-CORE AUDIT RESULTS (BTC, ETH, SOL, AVAX, ADA - DOGE EXCLUDED)") print("=" * 80) print(f"2-Year Seed: Final=${sum_2yr['final_equity']:>10,.2f} | CAGR={sum_2yr['cagr']:>5.2f}% | MaxDD={sum_2yr['max_dd']:>5.2f}% | Calmar={sum_2yr['calmar']:>4.2f} | PF={sum_2yr['pf']:>4.2f} | Trades={sum_2yr['total_trades']}") print(f"5-Year Staged BNB: Final=${sum_5yr_staged['final_equity']:>10,.2f} | CAGR={sum_5yr_staged['cagr']:>5.2f}% | MaxDD={sum_5yr_staged['max_dd']:>5.2f}% | Calmar={sum_5yr_staged['calmar']:>4.2f} | PF={sum_5yr_staged['pf']:>4.2f} | Trades={sum_5yr_staged['total_trades']}") print(f"5-Year Pure Seed: Final=${sum_5yr_seed['final_equity']:>10,.2f} | CAGR={sum_5yr_seed['cagr']:>5.2f}% | MaxDD={sum_5yr_seed['max_dd']:>5.2f}% | Calmar={sum_5yr_seed['calmar']:>4.2f} | PF={sum_5yr_seed['pf']:>4.2f} | Trades={sum_5yr_seed['total_trades']}") print("=" * 80) # 2-Year Asset Breakdown print("\n--- 2-YEAR PENTA-CORE ASSET BREAKDOWN ---") for s in SYMBOLS: st = [t for t in trades_2yr if t['symbol'] == s] sw = [t for t in st if t['is_win']] sl = [t for t in st if not t['is_win']] s_pnl = sum(t['net_pnl'] for t in st) s_gw = sum(t['net_pnl'] for t in sw) s_gl = abs(sum(t['net_pnl'] for t in sl)) spf = s_gw / s_gl if s_gl > 0 else 0.0 swr = len(sw) / len(st) * 100 if st else 0.0 print(f"{s:5} | Trades: {len(st):3} | Win Rate: {swr:5.1f}% | Net PnL: ${s_pnl:10,.2f} | PF: {spf:4.2f}")