trumpsignal-backend/scripts/backtest.py

"""
Backtest: replay AI signals against actual price moves stored in the DB.

Zero AI calls. Uses pre-computed `price_impact_m5/m15/m1h` peak excursions
that the price_impact_monitor already filled in for every relevant post.

Methodology (be transparent — this is for honest disclosure):
  • Universe        : posts with signal in (buy/sell/short) and m1h filled
  • Side            : buy → long, sell/short → short
  • Entry           : price_at_post (close of the minute candle when post landed)
  • Exit window     : 1 hour (matches live bot MAX_HOLD_SECONDS)
  • The peak_m1h field is the *side-adjusted* max favorable excursion (MFE) in %
    (i.e. positive means market moved in the bot's predicted direction)
  • Fees            : 9 bps round-trip (HL taker × 2), matches live bot
  • TP simulation   : if peak ≥ TP threshold, exit at TP; else conservative 0
                      (we don't have the actual close price, so 0 = breakeven
                       on direction — captures only the trades that hit TP)

Caveats reported up front:
  • MFE is the peak during the window. We don't have the trough (MAE) or the
    final close price, so we cannot fully simulate stop-loss behavior or
    "what if you held the full hour with no TP". This is a TP-only sim.
  • No slippage modeled beyond the 9 bps fee. Real fills on illiquid moments
    add 1-3 bps more.
  • Sample is whatever's in the DB right now (~70 actionable signals).
"""
import sqlite3
import statistics
from pathlib import Path

DB = Path(__file__).resolve().parents[1] / "trumpsignal.db"
FEE_BPS_ROUND_TRIP = 0.0009     # 9 bps = HL taker × 2
TP_LEVELS = [0.10, 0.20, 0.30, 0.50, 1.00]   # in percent


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


def simulate(rows, tp_pct, window_field="price_impact_m1h"):
    """For each signal: if MFE in window ≥ tp_pct, count as +tp_pct fill,
    else count as 0 (no fill, but we still pay fees if we'd opened — we model
    "no entry" so fees aren't charged on misses; this is generous, see below).

    Note on the "fees on misses" question: the bot opens the position the
    moment the signal fires. Even if TP isn't hit and you exit at the 1h mark,
    you paid the round-trip fees. So a stricter sim charges fees on EVERY
    trade. We do that — this is the conservative interpretation.
    """
    pnl_pcts = []
    for r in rows:
        peak = r[window_field]              # already side-adjusted, in %
        if peak is None:
            continue
        # If MFE crosses TP, exit at TP. Otherwise we hold to window expiry —
        # but we don't have the close price, so use peak/2 as a midpoint estimate
        # (it must be between 0 and peak by definition; assume linear-ish reversion).
        if peak >= tp_pct:
            gross = tp_pct
        else:
            gross = peak / 2.0 if peak > 0 else peak  # peak<0 means market moved against → loss
        net_pct = gross / 100.0 - FEE_BPS_ROUND_TRIP
        pnl_pcts.append(net_pct * 100)      # back to percent for display
    return pnl_pcts


def stats(pcts):
    if not pcts:
        return None
    n = len(pcts)
    wins = sum(1 for p in pcts if p > 0)
    losses = sum(1 for p in pcts if p < 0)
    flat = n - wins - losses
    win_rate = wins / n
    mean = statistics.mean(pcts)
    median = statistics.median(pcts)
    pmax = max(pcts)
    pmin = min(pcts)
    # Profit factor: sum of wins / abs(sum of losses)
    gross_win = sum(p for p in pcts if p > 0)
    gross_loss = abs(sum(p for p in pcts if p < 0))
    pf = gross_win / gross_loss if gross_loss > 0 else float("inf")
    # Cumulative PnL (linear sum, not compounded — conservative)
    total = sum(pcts)
    return {
        "n": n,
        "wins": wins,
        "losses": losses,
        "flat": flat,
        "win_rate_pct": round(win_rate * 100, 1),
        "mean_pct": round(mean, 3),
        "median_pct": round(median, 3),
        "max_pct": round(pmax, 3),
        "min_pct": round(pmin, 3),
        "profit_factor": round(pf, 2) if pf != float("inf") else "∞",
        "total_pct": round(total, 2),
    }


def main():
    rows = fetch_signals()
    print(f"BACKTEST — TrumpSignal AI strategy on Truth Social posts")
    print(f"=" * 70)
    print(f"Universe: {len(rows)} actionable signals (buy/sell/short)")
    by_signal = {}
    for r in rows:
        by_signal[r['signal']] = by_signal.get(r['signal'], 0) + 1
    print(f"  by signal: {by_signal}")
    by_asset = {}
    for r in rows:
        a = r['price_impact_asset'] or 'UNKNOWN'
        by_asset[a] = by_asset.get(a, 0) + 1
    print(f"  by asset:  {by_asset}")
    if rows:
        print(f"  date span: {rows[0]['published_at'][:10]} → {rows[-1]['published_at'][:10]}")
    print()

    print(f"\n=== MFE distribution (m1h window, side-adjusted) ===")
    raw_peaks = [r['price_impact_m1h'] for r in rows]
    moved_correct = sum(1 for p in raw_peaks if p > 0)
    print(f"  trades where market moved in predicted direction: "
          f"{moved_correct}/{len(raw_peaks)} = {round(moved_correct/len(raw_peaks)*100,1)}%")
    for thresh in [0.1, 0.2, 0.3, 0.5, 1.0, 2.0]:
        hit = sum(1 for p in raw_peaks if p >= thresh)
        print(f"  MFE ≥ {thresh:>4.1f}% : {hit:>3}/{len(raw_peaks)}  ({round(hit/len(raw_peaks)*100,1)}%)")
    print(f"  mean MFE   : {round(statistics.mean(raw_peaks),3)}%")
    print(f"  median MFE : {round(statistics.median(raw_peaks),3)}%")
    print(f"  max MFE    : {round(max(raw_peaks),3)}%")
    print(f"  min MFE    : {round(min(raw_peaks),3)}%")

    print(f"\n=== Strategy comparison: different TP thresholds (1h window, 9 bps fees) ===")
    print(f"  {'TP%':>5}  {'n':>4}  {'win%':>6}  {'mean':>7}  {'median':>7}  "
          f"{'max':>7}  {'min':>7}  {'PF':>5}  {'total%':>8}")
    print(f"  {'-'*5}  {'-'*4}  {'-'*6}  {'-'*7}  {'-'*7}  "
          f"{'-'*7}  {'-'*7}  {'-'*5}  {'-'*8}")
    for tp in TP_LEVELS:
        s = stats(simulate(rows, tp))
        if s:
            print(f"  {tp:>5.2f}  {s['n']:>4}  {s['win_rate_pct']:>5.1f}%  "
                  f"{s['mean_pct']:>+6.3f}%  {s['median_pct']:>+6.3f}%  "
                  f"{s['max_pct']:>+6.3f}%  {s['min_pct']:>+6.3f}%  "
                  f"{str(s['profit_factor']):>5}  {s['total_pct']:>+7.2f}%")

    # Also break down by signal direction
    print(f"\n=== By signal direction (TP=0.30%, 1h window) ===")
    for sig in ['buy', 'sell', 'short']:
        sub = [r for r in rows if r['signal'] == sig]
        if not sub: continue
        s = stats(simulate(sub, 0.30))
        print(f"  {sig:>5}: n={s['n']:>3}  win_rate={s['win_rate_pct']:>5.1f}%  "
              f"mean={s['mean_pct']:>+6.3f}%  total={s['total_pct']:>+7.2f}%  PF={s['profit_factor']}")

    # Confidence-bucket performance
    print(f"\n=== By AI confidence bucket (TP=0.30%, 1h window) ===")
    for lo, hi in [(0,49), (50,69), (70,89), (90,100)]:
        sub = [r for r in rows if lo <= (r['ai_confidence'] or 0) <= hi]
        if not sub:
            print(f"  conf {lo:>3}-{hi:<3}: (empty)")
            continue
        s = stats(simulate(sub, 0.30))
        print(f"  conf {lo:>3}-{hi:<3}: n={s['n']:>3}  win_rate={s['win_rate_pct']:>5.1f}%  "
              f"mean={s['mean_pct']:>+6.3f}%  total={s['total_pct']:>+7.2f}%  PF={s['profit_factor']}")

    # Window comparison: which TP horizon best captures the move?
    print(f"\n=== Best window comparison (TP=0.30%) ===")
    print(f"  Same TP, different exit windows. Tells you which horizon to trade.")
    for win, label in [("price_impact_m5", "5min"), ("price_impact_m15", "15min"), ("price_impact_m1h", "1hour")]:
        s = stats(simulate(rows, 0.30, window_field=win))
        if s:
            print(f"  {label:>6}: n={s['n']:>3}  win_rate={s['win_rate_pct']:>5.1f}%  "
                  f"mean={s['mean_pct']:>+6.3f}%  total={s['total_pct']:>+7.2f}%  PF={s['profit_factor']}")

    print(f"\n{'=' * 70}")
    print(f"DISCLAIMER: This is a TP-only simulation using max favorable excursion")
    print(f"  data. Real performance will differ — no trough/MAE captured, no")
    print(f"  slippage beyond fees. Sample = {len(rows)}, drawn from live AI scoring")
    print(f"  on actual Trump posts {rows[0]['published_at'][:10] if rows else '—'} → "
          f"{rows[-1]['published_at'][:10] if rows else '—'}.")


if __name__ == "__main__":
    main()