trumpsignal-backend/scripts/sweep_500.py

"""
Parameter sweep for a $500 starting account.

Finds (confidence_floor, leverage, tp%, sl%, margin_per_trade) combos that
maximize expected return while keeping max drawdown bounded.

Uses the v4 buy/short signals already in the DB (sell excluded — semantic
bug). For each signal we know `price_impact_m1h` = max favorable excursion
in % over the next hour (already side-adjusted). Sim:
  • peak >= tp_pct  → TP hit, exit at +tp
  • peak < 0        → SL hit (price went against us — pessimistic but safe)
  • else            → exit at peak/2 (held until window close, conservative)
  • fees: 9 bps round-trip on notional

Metrics computed per combo:
  total_pnl_usd  total profit on the $500 over the sample period
  max_dd_usd     biggest peak-to-trough drawdown in $
  max_dd_pct     same as % of starting capital
  sharpe_ish     mean / std of per-trade PnL  (rough Sharpe proxy)
  worst_trade    biggest single-trade loss in $
  n_trades       how many trades fired in this combo
"""
import sqlite3
import statistics
from itertools import product
from pathlib import Path

DB = Path(__file__).resolve().parents[1] / "trumpsignal.db"
CAPITAL = 500.0
FEE_BPS_RT = 0.0009


def fetch_signals(conf_floor: int):
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row
    rows = con.execute("""
        SELECT id, signal, ai_confidence, price_impact_m1h, published_at
          FROM posts
         WHERE signal IN ('buy', 'short')
           AND price_at_post IS NOT NULL
           AND price_impact_m1h IS NOT NULL
           AND ai_confidence >= ?
         ORDER BY published_at
    """, (conf_floor,)).fetchall()
    con.close()
    return [dict(r) for r in rows]


def simulate(signals, margin: float, leverage: int, tp_pct: float, sl_pct: float):
    """One sim of all signals with one parameter set. Returns trade-by-trade
    PnLs in $ (positive = win, negative = loss)."""
    notional = margin * leverage
    pnls: list[float] = []
    for s in signals:
        peak = s["price_impact_m1h"]            # already side-adjusted, %
        if peak >= tp_pct:
            gross = tp_pct
        elif peak < 0:
            gross = -sl_pct                     # pessimistic: assume SL hit
        else:
            gross = peak / 2                    # conservative midpoint
        # Apply fees on notional
        net_pct = gross / 100 - FEE_BPS_RT
        # Convert to $ on this single trade
        pnls.append(notional * net_pct)
    return pnls


def metrics(pnls: list[float]) -> dict:
    n = len(pnls)
    if n == 0:
        return {"n": 0, "total": 0.0, "max_dd": 0.0, "max_dd_pct": 0.0,
                "sharpe": 0.0, "worst": 0.0, "win_rate": 0.0}
    total = sum(pnls)
    wins = sum(1 for p in pnls if p > 0)
    # Running equity for drawdown
    equity = CAPITAL
    peak = equity
    max_dd = 0.0
    for p in pnls:
        equity += p
        if equity > peak:
            peak = equity
        dd = peak - equity
        if dd > max_dd:
            max_dd = dd
    sharpe = (statistics.mean(pnls) / statistics.stdev(pnls)
              if n > 1 and statistics.stdev(pnls) > 0 else 0.0)
    return {
        "n": n,
        "total": round(total, 2),
        "max_dd": round(max_dd, 2),
        "max_dd_pct": round(max_dd / CAPITAL * 100, 1),
        "sharpe": round(sharpe, 2),
        "worst": round(min(pnls), 2),
        "win_rate": round(wins / n * 100, 1),
    }


def main():
    print(f"$500 PARAMETER SWEEP")
    print("=" * 90)
    print(f"Sample: v4 buy/short signals from local DB, peak-MFE-based realistic sim")
    print(f"Sim model: TP if peak≥TP, SL if peak<0, else peak/2 (conservative)")
    print()

    # Sweep grid
    confs    = [60, 70, 75, 80, 85]
    margins  = [25, 50, 100]                   # $ per trade
    levs     = [5, 10, 15, 20]
    tps      = [0.5, 1.0, 1.5, 2.0, 3.0]      # %
    sls      = [0.5, 1.0, 1.5, 2.0]            # %

    results = []
    for conf in confs:
        sigs = fetch_signals(conf)
        if not sigs:
            continue
        for margin, lev, tp, sl in product(margins, levs, tps, sls):
            # Skip degenerate combos (tp<sl with low conviction signals)
            if margin * lev > CAPITAL * 5:    # cap notional at 5× capital total
                continue
            pnls = simulate(sigs, margin, lev, tp, sl)
            m = metrics(pnls)
            # ROI on the trading capital (not notional)
            m.update({
                "conf": conf, "margin": margin, "lev": lev,
                "tp": tp, "sl": sl,
                "roi_pct": round(m["total"] / CAPITAL * 100, 1),
                "max_loss_per_trade": round(margin * lev * (sl/100 + FEE_BPS_RT), 2),
            })
            results.append(m)

    # Filter to "realistic risk profile" first
    # Worst single trade <= $80 (16% of capital)
    # Max drawdown <= $150 (30% of capital)
    # At least 5 trades
    candidates = [
        r for r in results
        if r["max_loss_per_trade"] <= 80
        and r["max_dd"] <= 150
        and r["n"] >= 5
        and r["total"] > 0          # actually profitable
    ]

    print(f"Combos tested:        {len(results)}")
    print(f"Profitable + safe:    {len(candidates)}")
    print()

    # Top 10 by total PnL
    print("=" * 90)
    print("TOP 10 BY TOTAL PROFIT (with risk filters applied)")
    print("=" * 90)
    print(f"{'conf':>4}  {'margin':>6}  {'lev':>3}  {'TP%':>4}  {'SL%':>4}  "
          f"{'n':>3}  {'win%':>5}  {'total$':>8}  {'ROI%':>6}  "
          f"{'maxDD$':>7}  {'DD%':>5}  {'maxLoss$':>8}  {'sharpe':>6}")
    for r in sorted(candidates, key=lambda x: -x["total"])[:10]:
        print(f"{r['conf']:>4}  ${r['margin']:>5}  {r['lev']:>3}×  "
              f"{r['tp']:>4.1f}  {r['sl']:>4.1f}  "
              f"{r['n']:>3}  {r['win_rate']:>4.1f}%  "
              f"${r['total']:>+6.2f}  {r['roi_pct']:>+5.1f}%  "
              f"${r['max_dd']:>5.2f}  {r['max_dd_pct']:>4.1f}%  "
              f"${r['max_loss_per_trade']:>6.2f}  {r['sharpe']:>+5.2f}")

    print()
    print("=" * 90)
    print("TOP 5 BY SHARPE (risk-adjusted return)")
    print("=" * 90)
    print(f"{'conf':>4}  {'margin':>6}  {'lev':>3}  {'TP%':>4}  {'SL%':>4}  "
          f"{'n':>3}  {'win%':>5}  {'total$':>8}  {'sharpe':>6}  {'maxDD%':>6}")
    for r in sorted(candidates, key=lambda x: -x["sharpe"])[:5]:
        print(f"{r['conf']:>4}  ${r['margin']:>5}  {r['lev']:>3}×  "
              f"{r['tp']:>4.1f}  {r['sl']:>4.1f}  "
              f"{r['n']:>3}  {r['win_rate']:>4.1f}%  "
              f"${r['total']:>+6.2f}  {r['sharpe']:>+5.2f}  "
              f"{r['max_dd_pct']:>5.1f}%")

    print()
    print("=" * 90)
    print("LOWEST DRAWDOWN COMBOS (conservative, prioritize survival)")
    print("=" * 90)
    print(f"{'conf':>4}  {'margin':>6}  {'lev':>3}  {'TP%':>4}  {'SL%':>4}  "
          f"{'n':>3}  {'total$':>8}  {'maxDD%':>6}  {'sharpe':>6}")
    for r in sorted(candidates, key=lambda x: (x["max_dd_pct"], -x["total"]))[:5]:
        print(f"{r['conf']:>4}  ${r['margin']:>5}  {r['lev']:>3}×  "
              f"{r['tp']:>4.1f}  {r['sl']:>4.1f}  "
              f"{r['n']:>3}  ${r['total']:>+6.2f}  "
              f"{r['max_dd_pct']:>5.1f}%  {r['sharpe']:>+5.2f}")


if __name__ == "__main__":
    main()