#!/usr/bin/env python3 """ swing_trader_v2.py — EWT + DeMark Swing Trading Agent v2 Eligio Peraza / Karina AI — VPR Capital V2 IMPROVEMENTS: 1. Macro regime filter (SPY SMA200 + VIX gate) 2. RSI(14) as mandatory gate 3. Stricter volume confirmation (1.5x avg) 4. Stricter price flip (3 bars) 5. Trend alignment filter (MACD confirmation) Modes: scan — Scan watchlist, print only ACTIONABLE setups with entry/exit analyze TICKER — Deep single-asset analysis backtest TICKER — Historical stress test with accuracy metrics """ import argparse import json import sys import warnings from datetime import datetime, timedelta import numpy as np import pandas as pd import yfinance as yf warnings.filterwarnings("ignore") # ─── CONFIG ────────────────────────────────────────────────────────────────── WATCHLIST = ["SPY", "QQQ", "IWB", "IWM", "BTC-USD", "ETH-USD", "EURUSD=X", "GBPUSD=X", "JPY=X"] RISK_PCT = 0.015 # 1.5% account risk per trade RR_MIN = 3.0 # minimum Reward:Risk ratio FIB_EXT = 1.618 # Fibonacci extension for TP2 PROB_THRESHOLD = 65 # RAISED from 60 → 65 # ─── DATA ──────────────────────────────────────────────────────────────────── def fetch(ticker, period="1y", interval="1d"): df = yf.download(ticker, period=period, interval=interval, progress=False, auto_adjust=True) if isinstance(df.columns, pd.MultiIndex): df.columns = df.columns.get_level_values(0) df.dropna(inplace=True) return df def fetch_weekly(t): return fetch(t, "2y", "1wk") def fetch_daily(t): return fetch(t, "1y", "1d") def fetch_4h(t): df = fetch(t, "60d", "1h") if df.empty: return df df = df.resample("4h").agg({"Open":"first","High":"max","Low":"min","Close":"last","Volume":"sum"}).dropna() return df # ─── INDICATORS ─────────────────────────────────────────────────────────────── def atr(df, period=14): try: h, l, c = df["High"], df["Low"], df["Close"] tr = pd.concat([h - l, (h - c.shift()).abs(), (l - c.shift()).abs()], axis=1).max(axis=1) val = tr.rolling(period).mean().iloc[-1] if pd.isna(val): val = (df["High"] - df["Low"]).tail(period).mean() return float(val) except Exception: return float((df["High"] - df["Low"]).mean()) def sma(df, n): return df["Close"].rolling(n).mean() def rsi(df, period=14): """Calculate RSI(14). Returns last RSI value.""" close = df["Close"] delta = close.diff() gain = delta.where(delta > 0, 0.0) loss = -delta.where(delta < 0, 0.0) avg_gain = gain.rolling(period).mean() avg_loss = loss.rolling(period).mean() rs = avg_gain / avg_loss rsi_val = 100 - (100 / (1 + rs)) return float(rsi_val.iloc[-1]) if not pd.isna(rsi_val.iloc[-1]) else 50.0 def macd(df, fast=12, slow=26, signal=9): """Calculate MACD. Returns (macd_line, signal_line, histogram) — last values.""" close = df["Close"] ema_fast = close.ewm(span=fast).mean() ema_slow = close.ewm(span=slow).mean() macd_line = ema_fast - ema_slow signal_line = macd_line.ewm(span=signal).mean() hist = macd_line - signal_line return float(macd_line.iloc[-1]), float(signal_line.iloc[-1]), float(hist.iloc[-1]) def td_sequential(df): close = df["Close"].values n = len(close) buy_counts = [] sell_counts = [] bc = sc = 0 for i in range(4, n): if close[i] < close[i-4]: bc += 1; sc = 0 elif close[i] > close[i-4]: sc += 1; bc = 0 else: bc = max(0, bc); sc = max(0, sc) buy_counts.append(bc) sell_counts.append(sc) buy_count = buy_counts[-1] if buy_counts else 0 sell_count = sell_counts[-1] if sell_counts else 0 recent_buy = buy_counts[-5:] if len(buy_counts) >= 5 else buy_counts recent_sell = sell_counts[-5:] if len(sell_counts) >= 5 else sell_counts completed_9_buy = any(v >= 9 for v in recent_buy) completed_9_sell = any(v >= 9 for v in recent_sell) completed_13_buy = any(v >= 13 for v in buy_counts[-15:]) completed_13_sell = any(v >= 13 for v in sell_counts[-15:]) recycled = buy_count > 13 or sell_count > 13 return { "buy_count": int(buy_count), "sell_count": int(sell_count), "completed_9_buy": completed_9_buy, "completed_9_sell": completed_9_sell, "completed_13_buy": completed_13_buy, "completed_13_sell": completed_13_sell, "recycled": recycled, } def price_flip(df, bars=3): """V2: Stricter price flip — requires 3 bars (was 2).""" c = df["Close"].values[-(bars+2):] if len(c) < bars + 2: return False # Check: previous bars were going one direction, last bars reversed prev_dir = 1 if c[1] > c[0] else -1 last_dir = 1 if c[-1] > c[-2] else -1 mid_consistent = all( (1 if c[i+1] > c[i] else -1) == last_dir for i in range(len(c)-bars-1, len(c)-1) ) return prev_dir != last_dir and mid_consistent def trend_phase(df_weekly, df_daily): wma20 = sma(df_weekly, 20).iloc[-1] wma50 = sma(df_weekly, 50).iloc[-1] dma20 = sma(df_daily, 20).iloc[-1] dma50 = sma(df_daily, 50).iloc[-1] dma20_prev = sma(df_daily, 20).iloc[-5] w_close = df_weekly["Close"].iloc[-1] d_close = df_daily["Close"].iloc[-1] if w_close > wma20 > wma50: w_phase = "impulse_up" elif w_close < wma20 < wma50: w_phase = "impulse_down" elif wma20 > wma50 and w_close < wma20: w_phase = "correction_in_uptrend" else: w_phase = "transition" d_slope = (dma20 - dma20_prev) / dma20_prev * 100 if dma20_prev else 0 if d_close > dma20 > dma50: d_phase = "uptrend" elif d_close < dma20 < dma50: d_phase = "downtrend" elif d_slope > 0.1: d_phase = "accumulation" elif d_slope < -0.1: d_phase = "distribution" else: d_phase = "sideways" if w_phase in ("impulse_up", "correction_in_uptrend") and d_phase in ("uptrend", "accumulation"): bias = "BULLISH" elif w_phase == "impulse_down" and d_phase in ("downtrend", "distribution"): bias = "BEARISH" else: bias = "NEUTRAL" return w_phase, d_phase, bias def fibonacci_levels(df, lookback=60): subset = df.tail(lookback) swing_low = subset["Low"].min() swing_high = subset["High"].max() rng = swing_high - swing_low if rng == 0: return swing_low, swing_high, swing_low, swing_low, swing_high fib_50 = swing_high - 0.500 * rng fib_618 = swing_high - 0.618 * rng fib_ext = swing_low + FIB_EXT * rng return swing_low, swing_high, fib_50, fib_618, fib_ext def support_resistance(df, lookback=30): subset = df.tail(lookback) resistance = subset["High"].rolling(5, center=True).max().dropna() support = subset["Low"].rolling(5, center=True).min().dropna() key_resistance = resistance.nlargest(3).mean() key_support = support.nsmallest(3).mean() return key_support, key_resistance # ─── V2: MACRO REGIME FILTER ───────────────────────────────────────────────── def macro_regime(df_spy_daily): """ V2 NEW: Determine macro regime. Returns: (regime, risk_on) - RISK_ON: SPY > SMA200 → safe to buy pullbacks - RISK_OFF: SPY < SMA200 → avoid longs, prefer shorts or cash - CAUTION: SPY near SMA200 (within 2%) """ if df_spy_daily.empty or len(df_spy_daily) < 200: return "UNKNOWN", True # default to allow trading sma200 = df_spy_daily["Close"].rolling(200).mean().iloc[-1] current = float(df_spy_daily["Close"].iloc[-1]) if pd.isna(sma200): return "UNKNOWN", True pct_from_sma200 = (current - sma200) / sma200 * 100 if pct_from_sma200 > 2.0: return "RISK_ON", True elif pct_from_sma200 < -2.0: return "RISK_OFF", False else: return "CAUTION", True # near SMA200, allow but with lower probability def vix_check(threshold=30.0): """ V2 NEW: Check VIX level. Returns: (vix_value, is_elevated) VIX > 30 = extreme fear → avoid new longs VIX 25-30 = elevated → reduce position size VIX < 25 = normal """ try: vix = yf.download("^VIX", period="5d", interval="1d", progress=False, auto_adjust=True) if isinstance(vix.columns, pd.MultiIndex): vix.columns = vix.columns.get_level_values(0) vix_val = float(vix["Close"].iloc[-1]) return vix_val, vix_val > threshold except Exception: return 20.0, False # default: assume normal # ─── V2: SIGNAL LOGIC (IMPROVED) ───────────────────────────────────────────── # Cache for SPY data and VIX to avoid re-fetching per ticker _spy_cache = None _vix_cache = None def get_spy_data(): global _spy_cache if _spy_cache is None: _spy_cache = yf.download("SPY", start="2019-01-01", end=datetime.now().strftime("%Y-%m-%d"), interval="1d", progress=False, auto_adjust=True) if isinstance(_spy_cache.columns, pd.MultiIndex): _spy_cache.columns = _spy_cache.columns.get_level_values(0) return _spy_cache def get_vix_level(): global _vix_cache if _vix_cache is None: _vix_cache = vix_check(30.0) return _vix_cache def build_signal(ticker, df_weekly, df_daily, df_4h, account, df_spy_override=None, vix_override=None): """ V2: Apply all logic gates with macro overlay. """ if df_daily.empty or df_weekly.empty: return None current_price = float(df_daily["Close"].iloc[-1]) atr14 = atr(df_daily) # ═══ V2 NEW: MACRO REGIME CHECK ═══ if df_spy_override is not None: spy_data = df_spy_override else: spy_data = get_spy_data() # For backtest: use SPY data up to current date if not spy_data.empty: spy_slice = spy_data[spy_data.index <= df_daily.index[-1]] regime, risk_on = macro_regime(spy_slice) else: regime, risk_on = "UNKNOWN", True # V2 NEW: VIX check if vix_override is not None: vix_val, vix_elevated = vix_override else: vix_val, vix_elevated = get_vix_level() # Gate 1 – Market Phase w_phase, d_phase, bias = trend_phase(df_weekly, df_daily) # Gate 2 – TD Sequential td_d = td_sequential(df_daily) td_4h = td_sequential(df_4h) if not df_4h.empty else td_d # Gate 3 – Fibonacci & S/R swing_low, swing_high, fib_50, fib_618, fib_ext = fibonacci_levels(df_daily) key_support, key_resistance = support_resistance(df_daily) # Gate 4 – Price Flip (V2: stricter — 3 bars) flip = price_flip(df_daily, bars=3) # Gate 5 – Volume Confirmation (V2: stricter — 1.5x avg) has_volume = True vol_ratio = 1.0 if "Volume" in df_daily.columns and df_daily["Volume"].sum() > 0: avg_vol = df_daily["Volume"].rolling(20).mean().iloc[-1] last_vol = df_daily["Volume"].iloc[-1] vol_ratio = last_vol / avg_vol if avg_vol > 0 else 1.0 has_volume = vol_ratio >= 1.5 # V2: raised from 0.9 to 1.5 # ═══ V2 NEW: RSI GATE ═══ rsi_val = rsi(df_daily, 14) rsi_long_zone = 25 <= rsi_val <= 45 # oversold but not free-falling rsi_short_zone = 55 <= rsi_val <= 75 # overbought but not parabolic # ═══ V2 NEW: MACD CONFIRMATION ═══ macd_line, macd_signal, macd_hist = macd(df_daily) # For longs: MACD histogram turning positive (momentum shift) macd_long_ok = macd_hist > 0 or (macd_line > macd_signal) # For shorts: MACD histogram turning negative macd_short_ok = macd_hist < 0 or (macd_line < macd_signal) # ─ Determine direction ────────────────────────────────────────────────── direction = None entry_score = 0 # V2: 8 gates total (was 6) long_cond = [ bias == "BULLISH", # Gate 1: EWT weekly phase d_phase in ("accumulation", "uptrend", "sideways"), # Gate 1b: daily phase td_d["completed_9_buy"] or td_4h["completed_9_buy"], # Gate 2: TD-9 exhaustion current_price <= fib_618 * 1.02, # Gate 3: Fib retracement zone current_price >= key_support * 0.98, # Gate 3b: structural support has_volume, # Gate 5: Volume surge (V2: no OR with flip) rsi_long_zone, # Gate 6 NEW: RSI in buy zone macd_long_ok, # Gate 7 NEW: MACD confirming ] short_cond = [ bias == "BEARISH", d_phase in ("distribution", "downtrend", "sideways"), td_d["completed_9_sell"] or td_4h["completed_9_sell"], current_price >= fib_50 * 0.98, current_price <= key_resistance * 1.02, has_volume, rsi_short_zone, macd_short_ok, ] long_score = sum(long_cond) short_score = sum(short_cond) # Weekly exhaustion override weekly_exhausted = w_phase == "impulse_up" and (td_sequential(df_weekly)["completed_13_buy"]) weekly_impulse_up = w_phase in ("impulse_up", "correction_in_uptrend") # MANDATORY hard gates fib_confluence_long = current_price <= fib_618 * 1.02 fib_confluence_short = current_price >= fib_50 * 0.98 # ═══ V2: MACRO VETO ═══ macro_block_long = (regime == "RISK_OFF" and vix_elevated) # Block longs in risk-off + high VIX macro_block_short = (regime == "RISK_ON") # Don't short in strong bull market # V2: Require 5/8 gates (was 4/6) + mandatory gates + macro check + price flip if (long_score >= 5 and not weekly_exhausted and fib_confluence_long and flip and rsi_long_zone and not macro_block_long): direction = "LONG" entry_score = long_score elif (short_score >= 5 and not weekly_impulse_up and fib_confluence_short and flip and rsi_short_zone and not macro_block_short): direction = "SHORT" entry_score = short_score if direction is None: return {"ticker": ticker, "direction": "NEUTRAL", "actionable": False, "current_price": current_price, "bias": bias, "td_daily": td_d, "recycled": td_d["recycled"] or td_4h["recycled"], "w_phase": w_phase, "d_phase": d_phase, "regime": regime, "vix": vix_val, "rsi": rsi_val} # ─ Entry / Exit levels ────────────────────────────────────────────────── if direction == "LONG": entry = round(current_price, 4) sl = round(min(swing_low, current_price - atr14 * 1.5), 4) risk = abs(entry - sl) tp1 = round(entry + risk * 3.0, 4) tp2 = round(entry + risk * 5.0, 4) else: entry = round(current_price, 4) sl = round(max(swing_high, current_price + atr14 * 1.5), 4) risk = abs(sl - entry) tp1 = round(entry - risk * 3.0, 4) tp2 = round(entry - risk * 5.0, 4) risk = abs(entry - sl) reward = abs(tp1 - entry) rr = round(reward / risk, 2) if risk > 0 else 0 if rr < RR_MIN: return {"ticker": ticker, "direction": "NEUTRAL", "actionable": False, "current_price": current_price, "bias": bias, "td_daily": td_d, "recycled": td_d["recycled"], "rr_rejected": True, "rr": rr, "w_phase": w_phase, "d_phase": d_phase, "regime": regime, "vix": vix_val, "rsi": rsi_val} # Position sizing volatility_mult = 1.5 if "BTC" in ticker or "ETH" in ticker else (1.2 if "USD" in ticker else 1.0) # V2: Reduce size if VIX elevated (25-30) or CAUTION regime if vix_val > 25 or regime == "CAUTION": volatility_mult *= 1.3 # reduce position by ~23% risk_adj = risk * volatility_mult shares = int((account * RISK_PCT) / risk_adj) if risk_adj else 0 pos_size = round(shares * entry, 2) # V2: Improved probability score (more granular) prob = 35 # lower base (was 40) prob += entry_score * 8 # per gate (was 10) if flip: prob += 5 if not td_d["recycled"]: prob += 5 if regime == "RISK_ON": prob += 5 # V2: macro bonus if rsi_long_zone or rsi_short_zone: prob += 5 # V2: RSI bonus if vol_ratio >= 2.0: prob += 3 # V2: volume spike bonus prob = min(prob, 95) recycled = td_d["recycled"] or td_4h["recycled"] return { "ticker": ticker, "direction": direction, "actionable": prob >= PROB_THRESHOLD, "current_price": current_price, "bias": bias, "w_phase": w_phase, "d_phase": d_phase, "atr": round(atr14, 4), "td_daily": td_d, "td_4h": td_4h, "price_flip": flip, "rsi": round(rsi_val, 1), "macd_hist": round(macd_hist, 4), "regime": regime, "vix": round(vix_val, 1) if vix_val else None, "key_support": round(key_support, 4), "key_resistance": round(key_resistance, 4), "fib_50": round(fib_50, 4), "fib_618": round(fib_618, 4), "fib_ext": round(fib_ext, 4), "entry": entry, "stop_loss": sl, "take_profit_1": tp1, "take_profit_2": tp2, "rr": rr, "shares": shares, "position_size": pos_size, "probability": prob, "recycled": recycled, "holding_days_target": "2-8 days", } # ─── OUTPUT FORMATTING ──────────────────────────────────────────────────────── SEP = "═" * 62 def fmt_price(p, ticker): if "JPY" in ticker: return f"{p:.3f}" if "USD" in ticker and "BTC" not in ticker and "ETH" not in ticker: return f"{p:.4f}" return f"{p:.2f}" def print_signal(sig, account): t = sig["ticker"] p = sig["current_price"] td = sig.get("td_daily", {}) print(f"\n{SEP}") print(f" 📊 ANALYSIS (V2) {t}") print(f"{SEP}") print(f"\n ── MARKET CONTEXT ──────────────────────────────────") print(f" Price: {fmt_price(p, t)}") print(f" ATR(14): {sig.get('atr', 'N/A')}") print(f" RSI(14): {sig.get('rsi', 'N/A')}") print(f" MACD Hist: {sig.get('macd_hist', 'N/A')}") print(f" Macro Regime: {sig.get('regime', 'N/A')}") print(f" VIX: {sig.get('vix', 'N/A')}") print(f" Overall Bias: {sig.get('bias', 'N/A')}") print(f" Weekly Trend: {sig.get('w_phase', 'N/A')}") print(f" Daily Trend: {sig.get('d_phase', 'N/A')}") print(f" Key Support: {fmt_price(sig.get('key_support', 0), t)}") print(f" Key Resistance: {fmt_price(sig.get('key_resistance', 0), t)}") print(f" TD Daily: Buy={td.get('buy_count',0)} Sell={td.get('sell_count',0)}") print(f" Price Flip (3b): {sig.get('price_flip', False)}") d = sig.get("direction", "NEUTRAL") if d == "NEUTRAL": print(f"\n No actionable setup at this time.") if sig.get("rr_rejected"): print(f" ❌ R:R of {sig.get('rr', 0):.2f} below minimum {RR_MIN:.1f}") else: arrow = "🟢 LONG" if d == "LONG" else "🔴 SHORT" print(f"\n Signal: {arrow}") print(f" 🎯 ENTRY: {fmt_price(sig['entry'], t)}") print(f" 🛑 STOP LOSS: {fmt_price(sig['stop_loss'], t)}") print(f" 💰 TP1: {fmt_price(sig['take_profit_1'], t)}") print(f" 💰 TP2: {fmt_price(sig['take_profit_2'], t)}") print(f" R:R: {sig['rr']:.2f}:1 | Prob: {sig['probability']}%") print(f" Size: {sig['shares']} units (${sig['position_size']:,.0f})") print(f"{SEP}\n") # ─── BACKTESTING ────────────────────────────────────────────────────────────── BACKTEST_START = "2020-01-01" def mode_backtest(ticker, years, account): ticker = ticker.upper() print(f"\n{SEP}") print(f" BACKTEST V2 — {ticker} (desde {BACKTEST_START})") print(f" Strategy: EWT + TD + Fib + RSI + MACD + Macro Regime") print(SEP) end_date = datetime.now().strftime("%Y-%m-%d") df_daily = yf.download(ticker, start=BACKTEST_START, end=end_date, interval="1d", progress=False, auto_adjust=True) df_weekly = yf.download(ticker, start="2019-01-01", end=end_date, interval="1wk", progress=False, auto_adjust=True) # V2: Download SPY for macro regime check during backtest df_spy = yf.download("SPY", start="2019-01-01", end=end_date, interval="1d", progress=False, auto_adjust=True) # V2: Download VIX for historical VIX check df_vix = yf.download("^VIX", start=BACKTEST_START, end=end_date, interval="1d", progress=False, auto_adjust=True) for df in [df_daily, df_weekly, df_spy, df_vix]: if isinstance(df.columns, pd.MultiIndex): df.columns = df.columns.get_level_values(0) df.dropna(inplace=True) if df_daily.empty: print(f" No data for {ticker}") return print(f" Data range: {df_daily.index[0].date()} → {df_daily.index[-1].date()}") print(f" Total bars: {len(df_daily)}") print(f" Scanning for setups...\n") trades = [] in_position = False entry_trade = None MIN_BARS = 80 signals_generated = 0 macro_blocked = 0 for i in range(MIN_BARS, len(df_daily)): hist_daily = df_daily.iloc[:i] hist_weekly = df_weekly[df_weekly.index <= hist_daily.index[-1]] # V2: Get historical VIX for this date current_date = hist_daily.index[-1] vix_slice = df_vix[df_vix.index <= current_date] if not vix_slice.empty: vix_val = float(vix_slice["Close"].iloc[-1]) vix_override = (vix_val, vix_val > 30.0) else: vix_override = (20.0, False) if in_position and entry_trade: curr = hist_daily.iloc[-1] bars_held = i - entry_trade["bar_in"] hit_sl = curr["Low"] <= entry_trade["stop_loss"] if entry_trade["direction"] == "LONG" else curr["High"] >= entry_trade["stop_loss"] hit_tp = curr["High"] >= entry_trade["take_profit_1"] if entry_trade["direction"] == "LONG" else curr["Low"] <= entry_trade["take_profit_1"] timeout = bars_held > 45 if hit_sl or hit_tp or timeout: exit_price = entry_trade["stop_loss"] if hit_sl else (entry_trade["take_profit_1"] if hit_tp else float(curr["Close"])) pnl_pct = (exit_price - entry_trade["entry"]) / entry_trade["entry"] if entry_trade["direction"] == "SHORT": pnl_pct = -pnl_pct entry_trade.update({ "exit_price": round(exit_price, 4), "exit_date": hist_daily.index[-1], "bars_held": bars_held, "pnl_pct": round(pnl_pct * 100, 2), "win": hit_tp, "exit_reason": "TP" if hit_tp else ("SL" if hit_sl else "TIMEOUT"), "swing_trade": bars_held <= 15, }) trades.append(entry_trade) in_position = False entry_trade = None continue if not in_position and len(hist_weekly) >= 20: try: sig = build_signal(ticker, hist_weekly, hist_daily, pd.DataFrame(), account, df_spy_override=df_spy, vix_override=vix_override) signals_generated += 1 if sig and sig.get("actionable"): entry_trade = { "bar_in": i, "entry_date": hist_daily.index[-1], "entry": sig["entry"], "stop_loss": sig["stop_loss"], "take_profit_1": sig["take_profit_1"], "direction": sig["direction"], "rr": sig["rr"], "probability": sig["probability"], "recycled": sig.get("recycled", False), "fib_confluence": hist_daily["Close"].iloc[-1] <= sig.get("fib_618", 0) * 1.02, "with_weekly_trend": sig["bias"] != "NEUTRAL", "td9_mature": sig["td_daily"].get("completed_9_buy") or sig["td_daily"].get("completed_9_sell"), "price_flip": sig.get("price_flip", False), "rsi_zone": sig.get("rsi", 50), "regime": sig.get("regime", "UNKNOWN"), } in_position = True except Exception: continue if not trades: print(" No trades were executed in this period.") print(f" (Signals evaluated: {signals_generated})") # Still show buy & hold for reference bh_start_price = float(df_daily["Close"].iloc[0]) bh_end_price = float(df_daily["Close"].iloc[-1]) start_date = pd.Timestamp(BACKTEST_START) end_date_ts = df_daily.index[-1] years_elapsed = max((end_date_ts - start_date).days / 365.25, 0.1) bh_total = (bh_end_price / bh_start_price) bh_cagr = (bh_total ** (1 / years_elapsed) - 1) * 100 print(f" Buy & Hold CAGR: {bh_cagr:+.1f}%/yr") print(f" Strategy: 0 trades = 0% return (cash)") print(f" ℹ️ Model was too selective — no setups passed all V2 gates") return # Calculate metrics (same as v1) wins = [t for t in trades if t["win"]] losses = [t for t in trades if not t["win"]] swing_trades = [t for t in trades if t.get("swing_trade", True)] win_rate = len(wins) / len(trades) * 100 if trades else 0 avg_win = np.mean([t["pnl_pct"] for t in wins]) if wins else 0 avg_loss = np.mean([t["pnl_pct"] for t in losses]) if losses else 0 gross_profit = sum(t["pnl_pct"] for t in wins) gross_loss = abs(sum(t["pnl_pct"] for t in losses)) profit_factor = gross_profit / gross_loss if gross_loss else 999.0 avg_duration = np.mean([t["bars_held"] for t in wins]) if wins else 0 equity = [100.0] for t in trades: equity.append(equity[-1] * (1 + t["pnl_pct"] / 100)) peak = equity[0] max_dd = 0 for e in equity: peak = max(peak, e) dd = (peak - e) / peak * 100 max_dd = max(max_dd, dd) recovery_factor = (equity[-1] - equity[0]) / max_dd if max_dd else 999.0 start_date = pd.Timestamp(BACKTEST_START) end_date_ts = df_daily.index[-1] years_elapsed = max((end_date_ts - start_date).days / 365.25, 0.1) total_return = (equity[-1] / 100.0) annualized_return = (total_return ** (1 / years_elapsed) - 1) * 100 annualized_dollar = account * (annualized_return / 100) bh_start_price = float(df_daily["Close"].iloc[0]) bh_end_price = float(df_daily["Close"].iloc[-1]) bh_total = (bh_end_price / bh_start_price) bh_cagr = (bh_total ** (1 / years_elapsed) - 1) * 100 bh_dollar = account * (bh_cagr / 100) alpha = annualized_return - bh_cagr pnl_series = pd.Series([t["pnl_pct"] for t in trades]) if len(pnl_series) > 1: vol = pnl_series.std() * np.sqrt(252 / max(np.mean([t["bars_held"] for t in trades]), 1)) sharpe = (annualized_return - 4.0) / vol if vol > 0 else 0.0 else: sharpe = 0.0 # Qualitative breakdowns with_trend = [t for t in trades if t.get("with_weekly_trend")] fib_trades = [t for t in trades if t.get("fib_confluence")] mature_9 = [t for t in trades if t.get("td9_mature")] flip_trades = [t for t in trades if t.get("price_flip")] recycled_trades = [t for t in trades if t.get("recycled")] riskon_trades = [t for t in trades if t.get("regime") == "RISK_ON"] def win_pct(subset): w = [t for t in subset if t["win"]] return f"{len(w)/len(subset)*100:.0f}%" if subset else "N/A" def status(val, goal_lo, goal_hi=None, higher_better=True): if goal_hi: ok = goal_lo <= val <= goal_hi else: ok = val >= goal_lo if higher_better else val <= goal_lo return "✅ Pasa" if ok else "⚠️ No pasa" print(f"\n ══ STRATEGY V2 ACCURACY METRICS ═══════════════════════") print(f" {'Metric':<22} {'Goal':<14} {'Result':<14} {'Status'}") print(f" {'-'*22} {'-'*14} {'-'*14} {'-'*6}") print(f" {'Win Rate':<22} {'35–50%':<14} {f'{win_rate:.1f}%':<14} {status(win_rate, 35, 50)}") print(f" {'Profit Factor':<22} {'> 2.0':<14} {f'{profit_factor:.2f}':<14} {status(profit_factor, 2.0)}") print(f" {'Avg Win Duration':<22} {'5–12 Days':<14} {f'{avg_duration:.1f} days':<14} {status(avg_duration, 5, 12)}") print(f" {'Max Drawdown':<22} {'< 15%':<14} {f'{max_dd:.1f}%':<14} {status(max_dd, 15, higher_better=False)}") print(f" {'Recovery Factor':<22} {'> 2.0':<14} {f'{recovery_factor:.1f}':<14} {status(recovery_factor, 2.0)}") ann_str = f"{annualized_return:+.1f}%/yr (${annualized_dollar:+,.0f})" print(f" {'Rendimiento Anualiz.':<22} {'> 8%/yr':<14} {ann_str:<14} {status(annualized_return, 8)}") bh_str = f"{bh_cagr:+.1f}%/yr (${bh_dollar:+,.0f})" alpha_icon = "✅ Supera B&H" if alpha >= 0 else ("🟡 Casi empata" if alpha >= -3 else "⚠️ Underperforma") print(f" {'Buy & Hold (CAGR)':<22} {'benchmark':<14} {bh_str:<14} {'—'}") print(f" {'Alpha vs. B&H':<22} {'> 0%':<14} {f'{alpha:+.1f}%/yr':<28} {alpha_icon}") print(f" {'Sharpe Ratio':<22} {'> 1.0':<14} {f'{sharpe:.2f}':<14} {status(sharpe, 1.0)}") print(f"\n Total trades: {len(trades)} | Wins: {len(wins)} | Losses: {len(losses)}") print(f" Swing trades (≤15d): {len(swing_trades)} ({len(swing_trades)/len(trades)*100:.0f}%)") print(f" Avg win: +{avg_win:.1f}% Avg loss: {avg_loss:.1f}%") print(f"\n ── QUALITATIVE BREAKDOWN ───────────────────────────────") print(f" With Weekly Trend: {len(with_trend)} trades → Win rate: {win_pct(with_trend)}") print(f" Fib Confluence: {len(fib_trades)} trades → Win rate: {win_pct(fib_trades)}") print(f" Mature TD-9 Signal: {len(mature_9)} trades → Win rate: {win_pct(mature_9)}") print(f" Price Flip (3-bar): {len(flip_trades)} trades → Win rate: {win_pct(flip_trades)}") print(f" Risk-On Regime: {len(riskon_trades)} trades → Win rate: {win_pct(riskon_trades)}") print(f" Recycled Count Risk: {len(recycled_trades)} trades → Win rate: {win_pct(recycled_trades)}") print(f"\n{SEP}") try: result = { "version": "v2", "ticker": ticker, "win_rate": win_rate, "profit_factor": profit_factor, "max_drawdown": max_dd, "cagr": annualized_return, "bh_cagr": bh_cagr, "alpha": alpha, "sharpe": sharpe, "total_trades": len(trades), "wins": len(wins), "losses": len(losses), } with open("/tmp/swing_backtest_v2.json", "w") as f: json.dump(result, f, indent=2, default=str) print(f" Results saved to /tmp/swing_backtest_v2.json\n") except Exception: pass # ─── MAIN ───────────────────────────────────────────────────────────────────── def mode_scan(account): now = datetime.now() print(f"\n{SEP}") print(f" SWING TRADER V2 — SCAN MODE") print(f" {now.strftime('%Y-%m-%d %H:%M')} | Account: ${account:,.0f}") print(SEP) actionable = [] for t in WATCHLIST: sys.stdout.write(f" Scanning {t}... ") sys.stdout.flush() try: sig = build_signal(t, fetch_weekly(t), fetch_daily(t), fetch_4h(t), account) if sig and sig["actionable"]: print(f"{'LONG ✅' if sig['direction']=='LONG' else 'SHORT 🔴'} — {sig['probability']}%") actionable.append(sig) else: bias = sig.get("bias", "N/A") if sig else "ERR" print(f"neutral") except Exception as e: print(f"error: {e}") print(f"\n{SEP}") if not actionable: print(" No actionable setups found.") else: for sig in actionable: t = sig["ticker"] d = "🟢 LONG" if sig["direction"]=="LONG" else "🔴 SHORT" print(f" {d} {t} @ {fmt_price(sig['entry'], t)} | SL: {fmt_price(sig['stop_loss'], t)} | TP1: {fmt_price(sig['take_profit_1'], t)} | R:R: {sig['rr']:.1f}:1 | Prob: {sig['probability']}%") print(SEP) def mode_analyze(ticker, account): ticker = ticker.upper() sig = build_signal(ticker, fetch_weekly(ticker), fetch_daily(ticker), fetch_4h(ticker), account) print_signal(sig, account) def main(): parser = argparse.ArgumentParser(description="Swing Trader V2 — EWT + DeMark + Macro") sub = parser.add_subparsers(dest="mode", required=True) sp = sub.add_parser("scan") sp.add_argument("--account", type=float, default=100000.0) ap = sub.add_parser("analyze") ap.add_argument("ticker") ap.add_argument("--account", type=float, default=100000.0) bp = sub.add_parser("backtest") bp.add_argument("ticker") bp.add_argument("years", type=int, nargs="?", default=3) bp.add_argument("--account", type=float, default=100000.0) args = parser.parse_args() if args.mode == "scan": mode_scan(args.account) elif args.mode == "analyze": mode_analyze(args.ticker, args.account) elif args.mode == "backtest": mode_backtest(args.ticker, args.years, args.account) if __name__ == "__main__": main()